Merge pull request #7055 from ellemouton/wtclientMigrations

watchtower: reduce AckedUpdate storage footprint
This commit is contained in:
Oliver Gugger authored and GitHub committed 2023-01-17 10:37:52 +01:00
commit 04447927ea
28 files changed
+4016 -193

No files matched your search

+2
View File
@@ -108,6 +108,7 @@ func TestMigrateRevocationLog(t *testing.T) {
beforeMigration,
afterMigration,
MigrateRevocationLog,
false,
)
})
if !success {
@@ -569,5 +570,6 @@ func BenchmarkMigration(b *testing.B) {
return MigrateRevocationLog(db)
},
false,
)
}
+6 -3
View File
@@ -90,7 +90,7 @@ func ApplyMigration(t *testing.T,
// supplied migration functions to take a db instance and construct their own
// database transactions.
func ApplyMigrationWithDb(t testing.TB, beforeMigration, afterMigration,
migrationFunc func(db kvdb.Backend) error) {
migrationFunc func(db kvdb.Backend) error, shouldFail bool) {
t.Helper()
@@ -106,8 +106,11 @@ func ApplyMigrationWithDb(t testing.TB, beforeMigration, afterMigration,
}
// Apply migration.
if err := migrationFunc(cdb); err != nil {
t.Fatalf("migrationFunc error: %v", err)
err = migrationFunc(cdb)
if shouldFail {
require.Error(t, err)
} else {
require.NoError(t, err)
}
// If there's no afterMigration, exit here.
@@ -305,6 +305,9 @@ certain large transactions](https://github.com/lightningnetwork/lnd/pull/7100).
using the address in question](
https://github.com/lightningnetwork/lnd/pull/7025)
* [Store AckedUpdates in a more compact
way](https://github.com/lightningnetwork/lnd/pull/7055)
## Pathfinding
* [Pathfinding takes capacity of edges into account to improve success
+8 -7
View File
@@ -341,27 +341,27 @@ func constructFunctionalOptions(includeSessions bool) (
var (
opts []wtdb.ClientSessionListOption
ackCounts = make(map[wtdb.SessionID]uint16)
committedUpdateCounts = make(map[wtdb.SessionID]uint16)
ackCounts = make(map[wtdb.SessionID]uint16)
)
if !includeSessions {
return opts, ackCounts, committedUpdateCounts
}
perAckedUpdate := func(s *wtdb.ClientSession, _ uint16,
_ wtdb.BackupID) {
perNumAckedUpdates := func(s *wtdb.ClientSession, id lnwire.ChannelID,
numUpdates uint16) {
ackCounts[s.ID]++
ackCounts[s.ID] += numUpdates
}
perCommittedUpdate := func(s *wtdb.ClientSession,
_ *wtdb.CommittedUpdate) {
u *wtdb.CommittedUpdate) {
committedUpdateCounts[s.ID]++
}
opts = []wtdb.ClientSessionListOption{
wtdb.WithPerAckedUpdate(perAckedUpdate),
wtdb.WithPerNumAckedUpdates(perNumAckedUpdates),
wtdb.WithPerCommittedUpdate(perCommittedUpdate),
}
@@ -438,7 +438,8 @@ func (c *WatchtowerClient) Policy(ctx context.Context,
// marshallTower converts a client registered watchtower into its corresponding
// RPC type.
func marshallTower(tower *wtclient.RegisteredTower, includeSessions bool,
ackCounts, pendingCounts map[wtdb.SessionID]uint16) *Tower {
ackCounts map[wtdb.SessionID]uint16,
pendingCounts map[wtdb.SessionID]uint16) *Tower {
rpcAddrs := make([]string, 0, len(tower.Addresses))
for _, addr := range tower.Addresses {
+4
View File
@@ -4,6 +4,8 @@ import (
"github.com/btcsuite/btclog"
"github.com/lightningnetwork/lnd/build"
"github.com/lightningnetwork/lnd/watchtower/lookout"
"github.com/lightningnetwork/lnd/watchtower/wtclient"
"github.com/lightningnetwork/lnd/watchtower/wtdb"
"github.com/lightningnetwork/lnd/watchtower/wtserver"
)
@@ -30,4 +32,6 @@ func UseLogger(logger btclog.Logger) {
log = logger
lookout.UseLogger(logger)
wtserver.UseLogger(logger)
wtclient.UseLogger(logger)
wtdb.UseLogger(logger)
}
+11 -9
View File
@@ -314,7 +314,9 @@ func New(config *Config) (*TowerClient, error) {
// determine the highest known commit height for each channel. This
// allows the client to reject backups that it has already processed for
// its active policy.
perUpdate := func(policy wtpolicy.Policy, id wtdb.BackupID) {
perUpdate := func(policy wtpolicy.Policy, chanID lnwire.ChannelID,
commitHeight uint64) {
// We only want to consider accepted updates that have been
// accepted under an identical policy to the client's current
// policy.
@@ -324,22 +326,22 @@ func New(config *Config) (*TowerClient, error) {
// Take the highest commit height found in the session's acked
// updates.
height, ok := c.chanCommitHeights[id.ChanID]
if !ok || id.CommitHeight > height {
c.chanCommitHeights[id.ChanID] = id.CommitHeight
height, ok := c.chanCommitHeights[chanID]
if !ok || commitHeight > height {
c.chanCommitHeights[chanID] = commitHeight
}
}
perAckedUpdate := func(s *wtdb.ClientSession, _ uint16,
id wtdb.BackupID) {
perMaxHeight := func(s *wtdb.ClientSession, chanID lnwire.ChannelID,
height uint64) {
perUpdate(s.Policy, id)
perUpdate(s.Policy, chanID, height)
}
perCommittedUpdate := func(s *wtdb.ClientSession,
u *wtdb.CommittedUpdate) {
perUpdate(s.Policy, u.BackupID)
perUpdate(s.Policy, u.BackupID.ChanID, u.BackupID.CommitHeight)
}
// Load all candidate sessions and towers from the database into the
@@ -366,7 +368,7 @@ func New(config *Config) (*TowerClient, error) {
candidateSessions, err := getTowerAndSessionCandidates(
cfg.DB, cfg.SecretKeyRing, activeSessionFilter, perActiveTower,
wtdb.WithPerAckedUpdate(perAckedUpdate),
wtdb.WithPerMaxHeight(perMaxHeight),
wtdb.WithPerCommittedUpdate(perCommittedUpdate),
)
if err != nil {
+8
View File
@@ -68,6 +68,14 @@ type DB interface {
FetchSessionCommittedUpdates(id *wtdb.SessionID) (
[]wtdb.CommittedUpdate, error)
// IsAcked returns true if the given backup has been backed up using
// the given session.
IsAcked(id *wtdb.SessionID, backupID *wtdb.BackupID) (bool, error)
// NumAckedUpdates returns the number of backups that have been
// successfully backed up using the given session.
NumAckedUpdates(id *wtdb.SessionID) (uint64, error)
// FetchChanSummaries loads a mapping from all registered channels to
// their channel summaries.
FetchChanSummaries() (wtdb.ChannelSummaries, error)
+511 -103
View File
@@ -6,10 +6,12 @@ import (
"fmt"
"math"
"net"
"sync"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/tlv"
"github.com/lightningnetwork/lnd/watchtower/blob"
)
@@ -18,14 +20,23 @@ var (
// tower-id -> reserved-session-key-index (uint32).
cSessionKeyIndexBkt = []byte("client-session-key-index-bucket")
// cChanSummaryBkt is a top-level bucket storing:
// channel-id -> encoded ClientChanSummary.
cChanSummaryBkt = []byte("client-channel-summary-bucket")
// cChanDetailsBkt is a top-level bucket storing:
// channel-id => cChannelSummary -> encoded ClientChanSummary.
// => cChanDBID -> db-assigned-id
cChanDetailsBkt = []byte("client-channel-detail-bucket")
// cChanDBID is a key used in the cChanDetailsBkt to store the
// db-assigned-id of a channel.
cChanDBID = []byte("client-channel-db-id")
// cChannelSummary is a key used in cChanDetailsBkt to store the encoded
// body of ClientChanSummary.
cChannelSummary = []byte("client-channel-summary")
// cSessionBkt is a top-level bucket storing:
// session-id => cSessionBody -> encoded ClientSessionBody
// => cSessionCommits => seqnum -> encoded CommittedUpdate
// => cSessionAcks => seqnum -> encoded BackupID
// => cSessionAckRangeIndex => db-chan-id => start -> end
cSessionBkt = []byte("client-session-bucket")
// cSessionBody is a sub-bucket of cSessionBkt storing only the body of
@@ -36,9 +47,13 @@ var (
// seqnum -> encoded CommittedUpdate.
cSessionCommits = []byte("client-session-commits")
// cSessionAcks is a sub-bucket of cSessionBkt storing:
// seqnum -> encoded BackupID.
cSessionAcks = []byte("client-session-acks")
// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing
// chan-id => start -> end
cSessionAckRangeIndex = []byte("client-session-ack-range-index")
// cChanIDIndexBkt is a top-level bucket storing:
// db-assigned-id -> channel-ID
cChanIDIndexBkt = []byte("client-channel-id-index")
// cTowerBkt is a top-level bucket storing:
// tower-id -> encoded Tower.
@@ -67,6 +82,10 @@ var (
// structure deviates from what is expected.
ErrCorruptClientSession = errors.New("client session corrupted")
// ErrCorruptChanDetails signals that the clients channel detail's
// on-disk structure deviates from what is expected.
ErrCorruptChanDetails = errors.New("channel details corrupted")
// ErrClientSessionAlreadyExists signals an attempt to reinsert a client
// session that has already been created.
ErrClientSessionAlreadyExists = errors.New(
@@ -114,6 +133,11 @@ var (
// ErrLastTowerAddr is an error returned when the last address of a
// watchtower is attempted to be removed.
ErrLastTowerAddr = errors.New("cannot remove last tower address")
// ErrNoRangeIndexFound is returned when there is no persisted
// range-index found for the given session ID to channel ID pair.
ErrNoRangeIndexFound = errors.New("no range index found for the " +
"given session-channel pair")
)
// NewBoltBackendCreator returns a function that creates a new bbolt backend for
@@ -154,6 +178,12 @@ func NewBoltBackendCreator(active bool, dbPath,
// wtclient.
type ClientDB struct {
db kvdb.Backend
// ackedRangeIndex is a map from session ID to channel ID to a
// RangeIndex which represents the backups that have been acked for that
// channel using that session.
ackedRangeIndex map[SessionID]map[lnwire.ChannelID]*RangeIndex
ackedRangeIndexMu sync.Mutex
}
// OpenClientDB opens the client database given the path to the database's
@@ -171,6 +201,9 @@ func OpenClientDB(db kvdb.Backend) (*ClientDB, error) {
clientDB := &ClientDB{
db: db,
ackedRangeIndex: make(
map[SessionID]map[lnwire.ChannelID]*RangeIndex,
),
}
err = initOrSyncVersions(clientDB, firstInit, clientDBVersions)
@@ -198,11 +231,12 @@ func OpenClientDB(db kvdb.Backend) (*ClientDB, error) {
func initClientDBBuckets(tx kvdb.RwTx) error {
buckets := [][]byte{
cSessionKeyIndexBkt,
cChanSummaryBkt,
cChanDetailsBkt,
cSessionBkt,
cTowerBkt,
cTowerIndexBkt,
cTowerToSessionIndexBkt,
cChanIDIndexBkt,
}
for _, bucket := range buckets {
@@ -388,6 +422,11 @@ func (c *ClientDB) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
return ErrUninitializedDB
}
chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
if chanIDIndexBkt == nil {
return ErrUninitializedDB
}
// Don't return an error if the watchtower doesn't exist to act
// as a NOP.
pubKeyBytes := pubKey.SerializeCompressed()
@@ -428,8 +467,9 @@ func (c *ClientDB) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
committedUpdateCount[s.ID]++
}
towerSessions, err := listTowerSessions(
towerID, sessions, towersToSessionsIndex,
towerSessions, err := c.listTowerSessions(
towerID, sessions, chanIDIndexBkt,
towersToSessionsIndex,
WithPerCommittedUpdate(perCommittedUpdate),
)
if err != nil {
@@ -692,12 +732,186 @@ func (c *ClientDB) CreateClientSession(session *ClientSession) error {
return err
}
sessionBkt, err := sessions.CreateBucket(session.ID[:])
if err != nil {
return err
}
// Finally, write the client session's body in the sessions
// bucket.
return putClientSessionBody(sessions, session)
return putClientSessionBody(sessionBkt, session)
}, func() {})
}
// 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))
}
// getRangeIndex checks the ClientDB's in-memory range index map to see if it
// has an entry for the given session and channel ID. If it does, this is
// returned, otherwise the range index is loaded from the DB. An optional db
// transaction parameter may be provided. If one is provided then it will be
// used to query the DB for the range index, otherwise, a new transaction will
// be created and used.
func (c *ClientDB) getRangeIndex(tx kvdb.RTx, sID SessionID,
chanID lnwire.ChannelID) (*RangeIndex, error) {
c.ackedRangeIndexMu.Lock()
defer c.ackedRangeIndexMu.Unlock()
if _, ok := c.ackedRangeIndex[sID]; !ok {
c.ackedRangeIndex[sID] = make(map[lnwire.ChannelID]*RangeIndex)
}
// If the in-memory range-index map already includes an entry for this
// session ID and channel ID pair, then return it.
if index, ok := c.ackedRangeIndex[sID][chanID]; ok {
return index, nil
}
// readRangeIndexFromBkt is a helper that is used to read in a
// RangeIndex structure from the passed in bucket and store it in the
// ackedRangeIndex map.
readRangeIndexFromBkt := func(rangesBkt kvdb.RBucket) (*RangeIndex,
error) {
// Create a new in-memory RangeIndex by reading in ranges from
// the DB.
rangeIndex, err := readRangeIndex(rangesBkt)
if err != nil {
return nil, err
}
c.ackedRangeIndex[sID][chanID] = rangeIndex
return rangeIndex, nil
}
// If a DB transaction is provided then use it to fetch the ranges
// bucket from the DB.
if tx != nil {
rangesBkt, err := getRangesReadBucket(tx, sID, chanID)
if err != nil {
return nil, err
}
return readRangeIndexFromBkt(rangesBkt)
}
// No DB transaction was provided. So create and use a new one.
var index *RangeIndex
err := kvdb.View(c.db, func(tx kvdb.RTx) error {
rangesBkt, err := getRangesReadBucket(tx, sID, chanID)
if err != nil {
return err
}
index, err = readRangeIndexFromBkt(rangesBkt)
return err
}, func() {})
if err != nil {
return nil, err
}
return index, nil
}
// getRangesReadBucket gets the range index bucket where the range index for the
// given session-channel pair is stored. If any sub-buckets along the way do not
// exist, then an error is returned. If the sub-buckets should be created
// instead, then use getRangesWriteBucket.
func getRangesReadBucket(tx kvdb.RTx, sID SessionID, chanID lnwire.ChannelID) (
kvdb.RBucket, error) {
sessions := tx.ReadBucket(cSessionBkt)
if sessions == nil {
return nil, ErrUninitializedDB
}
chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return nil, ErrUninitializedDB
}
sessionBkt := sessions.NestedReadBucket(sID[:])
if sessionsBkt == nil {
return nil, ErrNoRangeIndexFound
}
// Get the DB representation of the channel-ID.
_, dbChanIDBytes, err := getDBChanID(chanDetailsBkt, chanID)
if err != nil {
return nil, err
}
sessionAckRanges := sessionBkt.NestedReadBucket(cSessionAckRangeIndex)
if sessionAckRanges == nil {
return nil, ErrNoRangeIndexFound
}
return sessionAckRanges.NestedReadBucket(dbChanIDBytes), nil
}
// getRangesWriteBucket gets the range index bucket where the range index for
// the given session-channel pair is stored. If any sub-buckets along the way do
// not exist, then they are created.
func getRangesWriteBucket(tx kvdb.RwTx, sID SessionID,
chanID lnwire.ChannelID) (kvdb.RwBucket, error) {
sessions := tx.ReadWriteBucket(cSessionBkt)
if sessions == nil {
return nil, ErrUninitializedDB
}
chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return nil, ErrUninitializedDB
}
sessionBkt, err := sessions.CreateBucketIfNotExists(sID[:])
if err != nil {
return nil, err
}
// Get the DB representation of the channel-ID.
_, dbChanIDBytes, err := getDBChanID(chanDetailsBkt, chanID)
if err != nil {
return nil, err
}
sessionAckRanges, err := sessionBkt.CreateBucketIfNotExists(
cSessionAckRangeIndex,
)
if err != nil {
return nil, err
}
return sessionAckRanges.CreateBucketIfNotExists(dbChanIDBytes)
}
// createSessionKeyIndexKey returns the identifier used in the
// session-key-index index, created as tower-id||blob-type.
//
@@ -760,13 +974,18 @@ func (c *ClientDB) ListClientSessions(id *TowerID,
return ErrUninitializedDB
}
chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
if chanIDIndexBkt == nil {
return ErrUninitializedDB
}
var err error
// If no tower ID is specified, then fetch all the sessions
// known to the db.
if id == nil {
clientSessions, err = listClientAllSessions(
sessions, opts...,
clientSessions, err = c.listClientAllSessions(
sessions, chanIDIndexBkt, opts...,
)
return err
}
@@ -777,8 +996,9 @@ func (c *ClientDB) ListClientSessions(id *TowerID,
return ErrUninitializedDB
}
clientSessions, err = listTowerSessions(
*id, sessions, towerToSessionIndex, opts...,
clientSessions, err = c.listTowerSessions(
*id, sessions, chanIDIndexBkt, towerToSessionIndex,
opts...,
)
return err
}, func() {
@@ -792,7 +1012,7 @@ func (c *ClientDB) ListClientSessions(id *TowerID,
}
// listClientAllSessions returns the set of all client sessions known to the db.
func listClientAllSessions(sessions kvdb.RBucket,
func (c *ClientDB) listClientAllSessions(sessions, chanIDIndexBkt kvdb.RBucket,
opts ...ClientSessionListOption) (map[SessionID]*ClientSession, error) {
clientSessions := make(map[SessionID]*ClientSession)
@@ -801,7 +1021,9 @@ func listClientAllSessions(sessions kvdb.RBucket,
// the CommittedUpdates and AckedUpdates on startup to resume
// committed updates and compute the highest known commit height
// for each channel.
session, err := getClientSession(sessions, k, opts...)
session, err := c.getClientSession(
sessions, chanIDIndexBkt, k, opts...,
)
if err != nil {
return err
}
@@ -819,7 +1041,7 @@ func listClientAllSessions(sessions kvdb.RBucket,
// listTowerSessions returns the set of all client sessions known to the db
// that are associated with the given tower id.
func listTowerSessions(id TowerID, sessionsBkt,
func (c *ClientDB) listTowerSessions(id TowerID, sessionsBkt, chanIDIndexBkt,
towerToSessionIndex kvdb.RBucket, opts ...ClientSessionListOption) (
map[SessionID]*ClientSession, error) {
@@ -834,7 +1056,9 @@ func listTowerSessions(id TowerID, sessionsBkt,
// the CommittedUpdates and AckedUpdates on startup to resume
// committed updates and compute the highest known commit height
// for each channel.
session, err := getClientSession(sessionsBkt, k, opts...)
session, err := c.getClientSession(
sessionsBkt, chanIDIndexBkt, k, opts...,
)
if err != nil {
return err
}
@@ -879,27 +1103,99 @@ func (c *ClientDB) FetchSessionCommittedUpdates(id *SessionID) (
return committedUpdates, nil
}
// FetchChanSummaries loads a mapping from all registered channels to their
// channel summaries.
func (c *ClientDB) FetchChanSummaries() (ChannelSummaries, error) {
var summaries map[lnwire.ChannelID]ClientChanSummary
// IsAcked returns true if the given backup has been backed up using the given
// session.
func (c *ClientDB) IsAcked(id *SessionID, backupID *BackupID) (bool, error) {
index, err := c.getRangeIndex(nil, *id, backupID.ChanID)
if errors.Is(err, ErrNoRangeIndexFound) {
return false, nil
} else if err != nil {
return false, err
}
return index.IsInIndex(backupID.CommitHeight), nil
}
// NumAckedUpdates returns the number of backups that have been successfully
// backed up using the given session.
func (c *ClientDB) NumAckedUpdates(id *SessionID) (uint64, error) {
var numAcked uint64
err := kvdb.View(c.db, func(tx kvdb.RTx) error {
chanSummaries := tx.ReadBucket(cChanSummaryBkt)
if chanSummaries == nil {
sessions := tx.ReadBucket(cSessionBkt)
if sessions == nil {
return ErrUninitializedDB
}
return chanSummaries.ForEach(func(k, v []byte) error {
var chanID lnwire.ChannelID
copy(chanID[:], k)
chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
if chanIDIndexBkt == nil {
return ErrUninitializedDB
}
var summary ClientChanSummary
err := summary.Decode(bytes.NewReader(v))
sessionBkt := sessions.NestedReadBucket(id[:])
if sessionsBkt == nil {
return nil
}
sessionAckRanges := sessionBkt.NestedReadBucket(
cSessionAckRangeIndex,
)
if sessionAckRanges == nil {
return nil
}
// Iterate over the channel ID's in the sessionAckRanges
// bucket.
return sessionAckRanges.ForEach(func(dbChanID, _ []byte) error {
// Get the range index for the session-channel pair.
chanIDBytes := chanIDIndexBkt.Get(dbChanID)
var chanID lnwire.ChannelID
copy(chanID[:], chanIDBytes)
index, err := c.getRangeIndex(tx, *id, chanID)
if err != nil {
return err
}
summaries[chanID] = summary
numAcked += index.NumInSet()
return nil
})
}, func() {
numAcked = 0
})
if err != nil {
return 0, err
}
return numAcked, nil
}
// FetchChanSummaries loads a mapping from all registered channels to their
// channel summaries.
func (c *ClientDB) FetchChanSummaries() (ChannelSummaries, error) {
var summaries map[lnwire.ChannelID]ClientChanSummary
err := kvdb.View(c.db, func(tx kvdb.RTx) error {
chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return ErrUninitializedDB
}
return chanDetailsBkt.ForEach(func(k, _ []byte) error {
chanDetails := chanDetailsBkt.NestedReadBucket(k)
if chanDetails == nil {
return ErrCorruptChanDetails
}
var chanID lnwire.ChannelID
copy(chanID[:], k)
summary, err := getChanSummary(chanDetails)
if err != nil {
return err
}
summaries[chanID] = *summary
return nil
})
@@ -922,23 +1218,50 @@ func (c *ClientDB) RegisterChannel(chanID lnwire.ChannelID,
sweepPkScript []byte) error {
return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
chanSummaries := tx.ReadWriteBucket(cChanSummaryBkt)
if chanSummaries == nil {
chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return ErrUninitializedDB
}
_, err := getChanSummary(chanSummaries, chanID)
switch {
// Summary already exists.
case err == nil:
chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID[:])
if chanDetails != nil {
// Channel is already registered.
return ErrChannelAlreadyRegistered
}
// Channel is not registered, proceed with registration.
case err == ErrChannelNotRegistered:
chanDetails, err := chanDetailsBkt.CreateBucket(chanID[:])
if err != nil {
return err
}
// Unexpected error.
default:
// Get the channel-id-index bucket.
indexBkt := tx.ReadWriteBucket(cChanIDIndexBkt)
if indexBkt == nil {
return ErrUninitializedDB
}
// Request the next unique id from the bucket.
nextSeq, err := indexBkt.NextSequence()
if err != nil {
return err
}
// Use BigSize encoding to encode the db-assigned index.
newIndex, err := writeBigSize(nextSeq)
if err != nil {
return err
}
// Add the new db-assigned ID to channel-ID pair.
err = indexBkt.Put(newIndex, chanID[:])
if err != nil {
return err
}
// Add the db-assigned ID to the channel's channel details
// bucket under the cChanDBID key.
err = chanDetails.Put(cChanDBID, newIndex)
if err != nil {
return err
}
@@ -946,7 +1269,7 @@ func (c *ClientDB) RegisterChannel(chanID lnwire.ChannelID,
SweepPkScript: sweepPkScript,
}
return putChanSummary(chanSummaries, chanID, &summary)
return putChanSummary(chanDetails, &summary)
}, func() {})
}
@@ -1031,7 +1354,7 @@ func (c *ClientDB) CommitUpdate(id *SessionID,
// eliminate serialization of full struct during CommitUpdate?
// Can also read/write directly to byes [:2] without migration.
session.SeqNum++
err = putClientSessionBody(sessions, session)
err = putClientSessionBody(sessionBkt, session)
if err != nil {
return err
}
@@ -1077,6 +1400,11 @@ func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16,
return ErrUninitializedDB
}
chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return ErrUninitializedDB
}
// We'll only load the ClientSession body for performance, since
// we primarily need to inspect its SeqNum and TowerLastApplied
// fields. The CommittedUpdates and AckedUpdates will be
@@ -1103,15 +1431,15 @@ func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16,
// also read/write directly to byes [2:4] without migration.
session.TowerLastApplied = lastApplied
// Can't fail because getClientSession succeeded.
sessionBkt := sessions.NestedReadWriteBucket(id[:])
// Write the client session with the updated last applied value.
err = putClientSessionBody(sessions, session)
err = putClientSessionBody(sessionBkt, session)
if err != nil {
return err
}
// Can't fail because of getClientSession succeeded.
sessionBkt := sessions.NestedReadWriteBucket(id[:])
// If the commits sub-bucket doesn't exist, there can't possibly
// be a corresponding committed update to remove.
sessionCommits := sessionBkt.NestedReadWriteBucket(
@@ -1145,25 +1473,24 @@ func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16,
return err
}
// Ensure that the session acks sub-bucket is initialized, so we
// can insert an entry.
sessionAcks, err := sessionBkt.CreateBucketIfNotExists(
cSessionAcks,
)
chanID := committedUpdate.BackupID.ChanID
height := committedUpdate.BackupID.CommitHeight
// Get the ranges write bucket before getting the range index to
// ensure that the session acks sub-bucket is initialized, so
// that we can insert an entry.
rangesBkt, err := getRangesWriteBucket(tx, *id, chanID)
if err != nil {
return err
}
// The session acks only need to track the backup id of the
// update, so we can discard the blob and hint.
var b bytes.Buffer
err = committedUpdate.BackupID.Encode(&b)
// Get the range index for the given session-channel pair.
index, err := c.getRangeIndex(tx, *id, chanID)
if err != nil {
return err
}
// Finally, insert the ack into the sessionAcks sub-bucket.
return sessionAcks.Put(seqNumBuf[:], b.Bytes())
return index.Add(height, rangesBkt)
}, func() {})
}
@@ -1196,9 +1523,15 @@ func getClientSessionBody(sessions kvdb.RBucket,
return &session, nil
}
// PerAckedUpdateCB describes the signature of a callback function that can be
// called for each of a session's acked updates.
type PerAckedUpdateCB func(*ClientSession, uint16, BackupID)
// PerMaxHeightCB describes the signature of a callback function that can be
// called for each channel that a session has updates for to communicate the
// maximum commitment height that the session has backed up for the channel.
type PerMaxHeightCB func(*ClientSession, lnwire.ChannelID, uint64)
// PerNumAckedUpdatesCB describes the signature of a callback function that can
// be called for each channel that a session has updates for to communicate the
// number of updates that the session has for the channel.
type PerNumAckedUpdatesCB func(*ClientSession, lnwire.ChannelID, uint16)
// PerCommittedUpdateCB describes the signature of a callback function that can
// be called for each of a session's committed updates (updates that the client
@@ -1213,9 +1546,15 @@ type ClientSessionListOption func(cfg *ClientSessionListCfg)
// ClientSessionListCfg defines various query parameters that will be used when
// querying the DB for client sessions.
type ClientSessionListCfg struct {
// PerAckedUpdate will, if set, be called for each of the session's
// acked updates.
PerAckedUpdate PerAckedUpdateCB
// PerNumAckedUpdates will, if set, be called for each of the session's
// channels to communicate the number of updates stored for that
// channel.
PerNumAckedUpdates PerNumAckedUpdatesCB
// PerMaxHeight will, if set, be called for each of the session's
// channels to communicate the highest commit height of updates stored
// for that channel.
PerMaxHeight PerMaxHeightCB
// PerCommittedUpdate will, if set, be called for each of the session's
// committed (un-acked) updates.
@@ -1227,11 +1566,22 @@ func NewClientSessionCfg() *ClientSessionListCfg {
return &ClientSessionListCfg{}
}
// WithPerAckedUpdate constructs a functional option that will set a call-back
// function to be called for each of a client's acked updates.
func WithPerAckedUpdate(cb PerAckedUpdateCB) ClientSessionListOption {
// WithPerMaxHeight constructs a functional option that will set a call-back
// function to be called for each of a session's channels to communicate the
// maximum commitment height that the session has stored for the channel.
func WithPerMaxHeight(cb PerMaxHeightCB) ClientSessionListOption {
return func(cfg *ClientSessionListCfg) {
cfg.PerAckedUpdate = cb
cfg.PerMaxHeight = cb
}
}
// WithPerNumAckedUpdates constructs a functional option that will set a
// call-back function to be called for each of a session's channels to
// communicate the number of updates that the session has stored for the
// channel.
func WithPerNumAckedUpdates(cb PerNumAckedUpdatesCB) ClientSessionListOption {
return func(cfg *ClientSessionListCfg) {
cfg.PerNumAckedUpdates = cb
}
}
@@ -1246,21 +1596,22 @@ func WithPerCommittedUpdate(cb PerCommittedUpdateCB) ClientSessionListOption {
// getClientSession loads the full ClientSession associated with the serialized
// session id. This method populates the CommittedUpdates, AckUpdates and Tower
// in addition to the ClientSession's body.
func getClientSession(sessions kvdb.RBucket, idBytes []byte,
opts ...ClientSessionListOption) (*ClientSession, error) {
func (c *ClientDB) getClientSession(sessionsBkt, chanIDIndexBkt kvdb.RBucket,
idBytes []byte, opts ...ClientSessionListOption) (*ClientSession,
error) {
cfg := NewClientSessionCfg()
for _, o := range opts {
o(cfg)
}
session, err := getClientSessionBody(sessions, idBytes)
session, err := getClientSessionBody(sessionsBkt, idBytes)
if err != nil {
return nil, err
}
// Can't fail because client session body has already been read.
sessionBkt := sessions.NestedReadBucket(idBytes)
sessionBkt := sessionsBkt.NestedReadBucket(idBytes)
// Pass the session's committed (un-acked) updates through the call-back
// if one is provided.
@@ -1273,7 +1624,10 @@ func getClientSession(sessions kvdb.RBucket, idBytes []byte,
// Pass the session's acked updates through the call-back if one is
// provided.
err = filterClientSessionAcks(sessionBkt, session, cfg.PerAckedUpdate)
err = c.filterClientSessionAcks(
sessionBkt, chanIDIndexBkt, session, cfg.PerMaxHeight,
cfg.PerNumAckedUpdates,
)
if err != nil {
return nil, err
}
@@ -1322,35 +1676,43 @@ func getClientSessionCommits(sessionBkt kvdb.RBucket, s *ClientSession,
// filterClientSessionAcks retrieves all acked updates for the session
// identified by the serialized session id and passes them to the provided
// call back if one is provided.
func filterClientSessionAcks(sessionBkt kvdb.RBucket, s *ClientSession,
cb PerAckedUpdateCB) error {
func (c *ClientDB) filterClientSessionAcks(sessionBkt,
chanIDIndexBkt kvdb.RBucket, s *ClientSession, perMaxCb PerMaxHeightCB,
perNumAckedUpdates PerNumAckedUpdatesCB) error {
if cb == nil {
if perMaxCb == nil && perNumAckedUpdates == nil {
return nil
}
sessionAcks := sessionBkt.NestedReadBucket(cSessionAcks)
if sessionAcks == nil {
sessionAcksRanges := sessionBkt.NestedReadBucket(cSessionAckRangeIndex)
if sessionAcksRanges == nil {
return nil
}
err := sessionAcks.ForEach(func(k, v []byte) error {
seqNum := byteOrder.Uint16(k)
return sessionAcksRanges.ForEach(func(dbChanID, _ []byte) error {
rangeBkt := sessionAcksRanges.NestedReadBucket(dbChanID)
if rangeBkt == nil {
return nil
}
var backupID BackupID
err := backupID.Decode(bytes.NewReader(v))
index, err := readRangeIndex(rangeBkt)
if err != nil {
return err
}
cb(s, seqNum, backupID)
chanIDBytes := chanIDIndexBkt.Get(dbChanID)
var chanID lnwire.ChannelID
copy(chanID[:], chanIDBytes)
if perMaxCb != nil {
perMaxCb(s, chanID, index.MaxHeight())
}
if perNumAckedUpdates != nil {
perNumAckedUpdates(s, chanID, uint16(index.NumInSet()))
}
return nil
})
if err != nil {
return err
}
return nil
}
// filterClientSessionCommits retrieves all committed updates for the session
@@ -1388,16 +1750,11 @@ func filterClientSessionCommits(sessionBkt kvdb.RBucket, s *ClientSession,
// putClientSessionBody stores the body of the ClientSession (everything but the
// CommittedUpdates and AckedUpdates).
func putClientSessionBody(sessions kvdb.RwBucket,
func putClientSessionBody(sessionBkt kvdb.RwBucket,
session *ClientSession) error {
sessionBkt, err := sessions.CreateBucketIfNotExists(session.ID[:])
if err != nil {
return err
}
var b bytes.Buffer
err = session.Encode(&b)
err := session.Encode(&b)
if err != nil {
return err
}
@@ -1410,15 +1767,19 @@ func putClientSessionBody(sessions kvdb.RwBucket,
func markSessionStatus(sessions kvdb.RwBucket, session *ClientSession,
status CSessionStatus) error {
sessionBkt, err := sessions.CreateBucketIfNotExists(session.ID[:])
if err != nil {
return err
}
session.Status = status
return putClientSessionBody(sessions, session)
return putClientSessionBody(sessionBkt, session)
}
// getChanSummary loads a ClientChanSummary for the passed chanID.
func getChanSummary(chanSummaries kvdb.RBucket,
chanID lnwire.ChannelID) (*ClientChanSummary, error) {
chanSummaryBytes := chanSummaries.Get(chanID[:])
func getChanSummary(chanDetails kvdb.RBucket) (*ClientChanSummary, error) {
chanSummaryBytes := chanDetails.Get(cChannelSummary)
if chanSummaryBytes == nil {
return nil, ErrChannelNotRegistered
}
@@ -1433,7 +1794,7 @@ func getChanSummary(chanSummaries kvdb.RBucket,
}
// putChanSummary stores a ClientChanSummary for the passed chanID.
func putChanSummary(chanSummaries kvdb.RwBucket, chanID lnwire.ChannelID,
func putChanSummary(chanDetails kvdb.RwBucket,
summary *ClientChanSummary) error {
var b bytes.Buffer
@@ -1442,7 +1803,7 @@ func putChanSummary(chanSummaries kvdb.RwBucket, chanID lnwire.ChannelID,
return err
}
return chanSummaries.Put(chanID[:], b.Bytes())
return chanDetails.Put(cChannelSummary, b.Bytes())
}
// getTower loads a Tower identified by its serialized tower id.
@@ -1473,3 +1834,50 @@ func putTower(towers kvdb.RwBucket, tower *Tower) error {
return towers.Put(tower.ID.Bytes(), b.Bytes())
}
// getDBChanID returns the db-assigned channel ID for the given real channel ID.
// It returns both the uint64 and byte representation.
func getDBChanID(chanDetailsBkt kvdb.RBucket, chanID lnwire.ChannelID) (uint64,
[]byte, error) {
chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
if chanDetails == nil {
return 0, nil, ErrChannelNotRegistered
}
idBytes := chanDetails.Get(cChanDBID)
if len(idBytes) == 0 {
return 0, nil, fmt.Errorf("no db-assigned ID found for "+
"channel ID %s", chanID)
}
id, err := readBigSize(idBytes)
if err != nil {
return 0, nil, err
}
return id, idBytes, nil
}
// writeBigSize will encode the given uint64 as a BigSize byte slice.
func writeBigSize(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
}
// readBigSize converts the given byte slice into a uint64 and assumes that the
// bytes slice is using BigSize encoding.
func readBigSize(b []byte) (uint64, error) {
r := bytes.NewReader(b)
i, err := tlv.ReadVarInt(r, &[8]byte{})
if err != nil {
return 0, err
}
return i, nil
}
+34 -32
View File
@@ -221,6 +221,26 @@ func (h *clientDBHarness) fetchSessionCommittedUpdates(id *wtdb.SessionID,
return updates
}
func (h *clientDBHarness) isAcked(id *wtdb.SessionID, backupID *wtdb.BackupID,
expErr error) bool {
h.t.Helper()
isAcked, err := h.db.IsAcked(id, backupID)
require.ErrorIs(h.t, err, expErr)
return isAcked
}
func (h *clientDBHarness) numAcked(id *wtdb.SessionID, expErr error) uint64 {
h.t.Helper()
numAcked, err := h.db.NumAckedUpdates(id)
require.ErrorIs(h.t, err, expErr)
return numAcked
}
// testCreateClientSession asserts various conditions regarding the creation of
// a new ClientSession. The test asserts:
// - client sessions can only be created if a session key index is reserved.
@@ -453,6 +473,7 @@ func testRemoveTower(h *clientDBHarness) {
}
h.insertSession(session, nil)
update := randCommittedUpdate(h.t, 1)
h.registerChan(update.BackupID.ChanID, nil, nil)
h.commitUpdate(&session.ID, update, nil)
// We should not be able to fully remove it from the database since
@@ -583,16 +604,6 @@ func testCommitUpdate(h *clientDBHarness) {
}, nil)
}
func perAckedUpdate(updates map[uint16]wtdb.BackupID) func(
_ *wtdb.ClientSession, seq uint16, id wtdb.BackupID) {
return func(_ *wtdb.ClientSession, seq uint16,
id wtdb.BackupID) {
updates[seq] = id
}
}
// testAckUpdate asserts the behavior of AckUpdate.
func testAckUpdate(h *clientDBHarness) {
const blobType = blob.TypeAltruistCommit
@@ -628,6 +639,8 @@ func testAckUpdate(h *clientDBHarness) {
// Commit to a random update at seqnum 1.
update1 := randCommittedUpdate(h.t, 1)
h.registerChan(update1.BackupID.ChanID, nil, nil)
lastApplied := h.commitUpdate(&session.ID, update1, nil)
require.Zero(h.t, lastApplied)
@@ -654,6 +667,7 @@ func testAckUpdate(h *clientDBHarness) {
// value is 1, since this was what was provided in the last successful
// ack.
update2 := randCommittedUpdate(h.t, 2)
h.registerChan(update2.BackupID.ChanID, nil, nil)
lastApplied = h.commitUpdate(&session.ID, update2, nil)
require.EqualValues(h.t, 1, lastApplied)
@@ -681,13 +695,16 @@ func (h *clientDBHarness) assertUpdates(id wtdb.SessionID,
expectedPending []wtdb.CommittedUpdate,
expectedAcked map[uint16]wtdb.BackupID) {
ackedUpdates := make(map[uint16]wtdb.BackupID)
_ = h.listSessions(
nil, wtdb.WithPerAckedUpdate(perAckedUpdate(ackedUpdates)),
)
committedUpates := h.fetchSessionCommittedUpdates(&id, nil)
checkCommittedUpdates(h.t, committedUpates, expectedPending)
checkAckedUpdates(h.t, ackedUpdates, expectedAcked)
committedUpdates := h.fetchSessionCommittedUpdates(&id, nil)
checkCommittedUpdates(h.t, committedUpdates, expectedPending)
// Check acked updates.
numAcked := h.numAcked(&id, nil)
require.EqualValues(h.t, len(expectedAcked), numAcked)
for _, backupID := range expectedAcked {
isAcked := h.isAcked(&id, &backupID, nil)
require.True(h.t, isAcked)
}
}
// checkCommittedUpdates asserts that the CommittedUpdates on session match the
@@ -707,21 +724,6 @@ func checkCommittedUpdates(t *testing.T, actualUpdates,
require.Equal(t, expUpdates, actualUpdates)
}
// checkAckedUpdates asserts that the AckedUpdates on a session match the
// expUpdates provided.
func checkAckedUpdates(t *testing.T, actualUpdates,
expUpdates map[uint16]wtdb.BackupID) {
// We promote nil expUpdates to an initialized map since the database
// should never return a nil map. This promotion is done purely out of
// convenience for the testing framework.
if expUpdates == nil {
expUpdates = make(map[uint16]wtdb.BackupID)
}
require.Equal(t, expUpdates, actualUpdates)
}
// TestClientDB asserts the behavior of a fresh client db, a reopened client db,
// and the mock implementation. This ensures that all databases function
// identically, especially in the negative paths.
+3 -3
View File
@@ -125,9 +125,9 @@ type dbObject interface {
Decode(io.Reader) error
}
// TestCodec serializes and deserializes wtdb objects in order to test that that
// the codec understands all of the required field types. The test also asserts
// that decoding an object into another results in an equivalent object.
// TestCodec serializes and deserializes wtdb objects in order to test that the
// codec understands all of the required field types. The test also asserts that
// decoding an object into another results in an equivalent object.
func TestCodec(tt *testing.T) {
var t *testing.T
+6
View File
@@ -4,6 +4,9 @@ import (
"github.com/btcsuite/btclog"
"github.com/lightningnetwork/lnd/build"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration1"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration2"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration3"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration4"
)
// log is a logger that is initialized with no output filters. This
@@ -28,6 +31,9 @@ func DisableLog() {
func UseLogger(logger btclog.Logger) {
log = logger
migration1.UseLogger(logger)
migration2.UseLogger(logger)
migration3.UseLogger(logger)
migration4.UseLogger(logger)
}
// logClosure is used to provide a closure over expensive logging operations so
+82
View File
@@ -0,0 +1,82 @@
package migration2
import (
"errors"
"github.com/lightningnetwork/lnd/kvdb"
)
var (
// cChanSummaryBkt is a top-level bucket storing:
// channel-id -> encoded ClientChanSummary.
cChanSummaryBkt = []byte("client-channel-summary-bucket")
// cChanDetailsBkt is a top-level bucket storing:
// channel-id => cChannelSummary -> encoded ClientChanSummary.
cChanDetailsBkt = []byte("client-channel-detail-bucket")
// cChannelSummary is a key used in cChanDetailsBkt to store the encoded
// body of ClientChanSummary.
cChannelSummary = []byte("client-channel-summary")
// ErrUninitializedDB signals that top-level buckets for the database
// have not been initialized.
ErrUninitializedDB = errors.New("db not initialized")
// ErrCorruptChanSummary signals that the clients channel summary's
// on-disk structure deviates from what is expected.
ErrCorruptChanSummary = errors.New("channel summary corrupted")
)
// MigrateClientChannelDetails creates a new channel-details bucket that uses
// channel IDs as sub-buckets where the channel summaries are moved to from the
// channel summary bucket. If the migration is successful then the channel
// summary bucket is deleted.
func MigrateClientChannelDetails(tx kvdb.RwTx) error {
log.Infof("Migrating the tower client db to move the channel " +
"summaries to the new channel-details bucket")
// Create the new top level cChanDetailsBkt.
chanDetailsBkt, err := tx.CreateTopLevelBucket(cChanDetailsBkt)
if err != nil {
return err
}
// Get the top-level channel summaries bucket.
chanSummaryBkt := tx.ReadWriteBucket(cChanSummaryBkt)
if chanSummaryBkt == nil {
return ErrUninitializedDB
}
// Iterate over the cChanSummaryBkt's keys. Each key is a channel-id.
// For each of these, create a new sub-bucket with this key in
// cChanDetailsBkt. In this sub-bucket, add the cChannelSummary key with
// the encoded ClientChanSummary as the value.
err = chanSummaryBkt.ForEach(func(chanID, summary []byte) error {
// Force the migration to fail if the summary is empty. This
// should never be the case, but it is added so that we can
// force the migration to fail in a test so that we can test
// that the db remains unaffected if a migration failure takes
// place.
if len(summary) == 0 {
return ErrCorruptChanSummary
}
// Create a new sub-bucket in the channel details bucket using
// this channel ID.
channelBkt, err := chanDetailsBkt.CreateBucket(chanID)
if err != nil {
return err
}
// Add the encoded channel summary in the new bucket under the
// channel-summary key.
return channelBkt.Put(cChannelSummary, summary)
})
if err != nil {
return err
}
// Now delete the cChanSummaryBkt from the DB.
return tx.DeleteTopLevelBucket(cChanSummaryBkt)
}
@@ -0,0 +1,124 @@
package migration2
import (
"encoding/binary"
"testing"
"github.com/lightningnetwork/lnd/channeldb/migtest"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/stretchr/testify/require"
)
var (
// pre is the expected data in the cChanSummaryBkt bucket before the
// migration.
pre = map[string]interface{}{
channelIDString(1): string([]byte{1, 2, 3}),
channelIDString(2): string([]byte{3, 4, 5}),
}
// pre should fail the migration due to no channel summary being found
// for a given channel ID.
preFailCorruptDB = map[string]interface{}{
channelIDString(1): string([]byte{1, 2, 3}),
channelIDString(2): "",
}
// post is the expected data after migration.
post = map[string]interface{}{
channelIDString(1): map[string]interface{}{
string(cChannelSummary): string([]byte{1, 2, 3}),
},
channelIDString(2): map[string]interface{}{
string(cChannelSummary): string([]byte{3, 4, 5}),
},
}
)
// TestMigrateClientChannelDetails tests that the MigrateClientChannelDetails
// function correctly moves channel-summaries from the channel-summaries bucket
// to the new channel-details bucket.
func TestMigrateClientChannelDetails(t *testing.T) {
t.Parallel()
tests := []struct {
name string
shouldFail bool
pre map[string]interface{}
post map[string]interface{}
}{
{
name: "migration ok",
shouldFail: false,
pre: pre,
post: post,
},
{
name: "fail due to corrupt db",
shouldFail: true,
pre: preFailCorruptDB,
post: nil,
},
{
name: "no channel summaries",
shouldFail: false,
pre: nil,
post: nil,
},
}
for _, test := range tests {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
// Before the migration we have a channel summary
// bucket.
before := func(tx kvdb.RwTx) error {
return migtest.RestoreDB(
tx, cChanSummaryBkt, test.pre,
)
}
// After the migration, we should have a new channel
// details bucket and no longer have a channel summary
// bucket.
after := func(tx kvdb.RwTx) error {
// If we expect our migration to fail, we
// expect our channel summary bucket to remain
// intact.
if test.shouldFail {
return migtest.VerifyDB(
tx, cChanSummaryBkt, test.pre,
)
}
// Otherwise, we expect the channel summary
// bucket to be deleted.
err := migtest.VerifyDB(
tx, cChanSummaryBkt, test.pre,
)
require.ErrorContains(t, err, "not found")
// We also expect the new channel details bucket
// to be present.
return migtest.VerifyDB(
tx, cChanDetailsBkt, test.post,
)
}
migtest.ApplyMigration(
t, before, after, MigrateClientChannelDetails,
test.shouldFail,
)
})
}
}
func channelIDString(id uint64) string {
var chanID ChannelID
binary.BigEndian.PutUint64(chanID[:], id)
return chanID.String()
}
+17
View File
@@ -0,0 +1,17 @@
package migration2
import "encoding/hex"
// ChannelID is a series of 32-bytes that uniquely identifies all channels
// within the network. The ChannelID is computed using the outpoint of the
// funding transaction (the txid, and output index). Given a funding output the
// ChannelID can be calculated by XOR'ing the big-endian serialization of the
// txid and the big-endian serialization of the output index, truncated to
// 2 bytes.
type ChannelID [32]byte
// String returns the string representation of the ChannelID. This is just the
// hex string encoding of the ChannelID itself.
func (c ChannelID) String() string {
return hex.EncodeToString(c[:])
}
+14
View File
@@ -0,0 +1,14 @@
package migration2
import (
"github.com/btcsuite/btclog"
)
// log is a logger that is initialized as disabled. This means the package will
// not perform any logging by default until a logger is set.
var log = btclog.Disabled
// UseLogger uses a specified Logger to output package logging info.
func UseLogger(logger btclog.Logger) {
log = logger
}
+104
View File
@@ -0,0 +1,104 @@
package migration3
import (
"bytes"
"encoding/binary"
"errors"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/tlv"
)
var (
// cChanDetailsBkt is a top-level bucket storing:
// channel-id => cChannelSummary -> encoded ClientChanSummary.
// => cChanDBID -> db-assigned-id
cChanDetailsBkt = []byte("client-channel-detail-bucket")
// cChanDBID is a key used in the cChanDetailsBkt to store the
// db-assigned-id of a channel.
cChanDBID = []byte("client-channel-db-id")
// cChanIDIndexBkt is a top-level bucket storing:
// db-assigned-id -> channel-ID
cChanIDIndexBkt = []byte("client-channel-id-index")
// cChannelSummary is a key used in cChanDetailsBkt to store the encoded
// body of ClientChanSummary.
cChannelSummary = []byte("client-channel-summary")
// ErrUninitializedDB signals that top-level buckets for the database
// have not been initialized.
ErrUninitializedDB = errors.New("db not initialized")
// ErrCorruptChanDetails signals that the clients channel detail's
// on-disk structure deviates from what is expected.
ErrCorruptChanDetails = errors.New("channel details corrupted")
byteOrder = binary.BigEndian
)
// MigrateChannelIDIndex adds a new channel ID index to the tower client db.
// This index is a mapping from db-assigned ID (a uint64 encoded using BigSize
// encoding) to real channel ID (32 bytes). This mapping will allow us to
// persist channel pointers with fewer bytes in the future.
func MigrateChannelIDIndex(tx kvdb.RwTx) error {
log.Infof("Migrating the tower client db to add a new channel ID " +
"index which stores a mapping from db-assigned ID to real " +
"channel ID")
// Create a new top-level bucket for the new index.
indexBkt, err := tx.CreateTopLevelBucket(cChanIDIndexBkt)
if err != nil {
return err
}
// Get the top-level channel-details bucket. The keys of this bucket
// are the real channel IDs.
chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return ErrUninitializedDB
}
// Iterate over the keys of the channel-details bucket.
return chanDetailsBkt.ForEach(func(chanID, _ []byte) error {
// Ask the db for a new, unique, ID for the index bucket.
nextSeq, err := indexBkt.NextSequence()
if err != nil {
return err
}
// Encode the sequence number using BigSize encoding.
var newIndex bytes.Buffer
err = tlv.WriteVarInt(&newIndex, nextSeq, &[8]byte{})
if err != nil {
return err
}
// Add the mapping from the db-assigned ID to the channel ID
// to the new index.
newIndexBytes := newIndex.Bytes()
err = indexBkt.Put(newIndexBytes, chanID)
if err != nil {
return err
}
chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID)
if chanDetails == nil {
return ErrCorruptChanDetails
}
// Here we ensure that the channel-details bucket includes a
// channel summary. The only reason we do this is so that we can
// simulate a migration fail in a test to ensure that a
// migration fail results in an untouched db.
chanSummaryBytes := chanDetails.Get(cChannelSummary)
if chanSummaryBytes == nil {
return ErrCorruptChanDetails
}
// In the channel-details sub-bucket for this channel, add the
// new DB-assigned ID for this channel under the cChanDBID key.
return chanDetails.Put(cChanDBID, newIndexBytes)
})
}
@@ -0,0 +1,149 @@
package migration3
import (
"bytes"
"testing"
"github.com/lightningnetwork/lnd/channeldb/migtest"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/tlv"
)
var (
// pre is the expected data in the cChanDetailsBkt bucket before the
// migration.
pre = map[string]interface{}{
channelIDString(100): map[string]interface{}{
string(cChannelSummary): string([]byte{1, 2, 3}),
},
channelIDString(222): map[string]interface{}{
string(cChannelSummary): string([]byte{4, 5, 6}),
},
}
// preFailCorruptDB should fail the migration due to no channel summary
// being found for a given channel ID.
preFailCorruptDB = map[string]interface{}{
channelIDString(100): map[string]interface{}{},
}
// post is the expected data in the new index after migration.
postIndex = map[string]interface{}{
indexToString(1): channelIDString(100),
indexToString(2): channelIDString(222),
}
// postDetails is the expected data in the cChanDetailsBkt bucket after
// the migration.
postDetails = map[string]interface{}{
channelIDString(100): map[string]interface{}{
string(cChannelSummary): string([]byte{1, 2, 3}),
string(cChanDBID): indexToString(1),
},
channelIDString(222): map[string]interface{}{
string(cChannelSummary): string([]byte{4, 5, 6}),
string(cChanDBID): indexToString(2),
},
}
)
// TestMigrateChannelIDIndex tests that the MigrateChannelIDIndex function
// correctly adds a new channel-id index to the DB and also correctly updates
// the existing channel-details bucket.
func TestMigrateChannelIDIndex(t *testing.T) {
t.Parallel()
tests := []struct {
name string
shouldFail bool
pre map[string]interface{}
postDetails map[string]interface{}
postIndex map[string]interface{}
}{
{
name: "migration ok",
shouldFail: false,
pre: pre,
postDetails: postDetails,
postIndex: postIndex,
},
{
name: "fail due to corrupt db",
shouldFail: true,
pre: preFailCorruptDB,
},
{
name: "no channel details",
shouldFail: false,
pre: nil,
postDetails: nil,
postIndex: nil,
},
}
for _, test := range tests {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
// Before the migration we have a details bucket.
before := func(tx kvdb.RwTx) error {
return migtest.RestoreDB(
tx, cChanDetailsBkt, test.pre,
)
}
// After the migration, we should have an untouched
// summary bucket and a new index bucket.
after := func(tx kvdb.RwTx) error {
// If the migration fails, the details bucket
// should be untouched.
if test.shouldFail {
if err := migtest.VerifyDB(
tx, cChanDetailsBkt, test.pre,
); err != nil {
return err
}
return nil
}
// Else, we expect an updated summary bucket
// and a new index bucket.
err := migtest.VerifyDB(
tx, cChanDetailsBkt, test.postDetails,
)
if err != nil {
return err
}
return migtest.VerifyDB(
tx, cChanIDIndexBkt, test.postIndex,
)
}
migtest.ApplyMigration(
t, before, after, MigrateChannelIDIndex,
test.shouldFail,
)
})
}
}
func indexToString(id uint64) string {
var newIndex bytes.Buffer
err := tlv.WriteVarInt(&newIndex, id, &[8]byte{})
if err != nil {
panic(err)
}
return newIndex.String()
}
func channelIDString(id uint64) string {
var chanID ChannelID
byteOrder.PutUint64(chanID[:], id)
return chanID.String()
}
+19
View File
@@ -0,0 +1,19 @@
package migration3
import (
"encoding/hex"
)
// ChannelID is a series of 32-bytes that uniquely identifies all channels
// within the network. The ChannelID is computed using the outpoint of the
// funding transaction (the txid, and output index). Given a funding output the
// ChannelID can be calculated by XOR'ing the big-endian serialization of the
// txid and the big-endian serialization of the output index, truncated to
// 2 bytes.
type ChannelID [32]byte
// String returns the string representation of the ChannelID. This is just the
// hex string encoding of the ChannelID itself.
func (c ChannelID) String() string {
return hex.EncodeToString(c[:])
}
+14
View File
@@ -0,0 +1,14 @@
package migration3
import (
"github.com/btcsuite/btclog"
)
// log is a logger that is initialized as disabled. This means the package will
// not perform any logging by default until a logger is set.
var log = btclog.Disabled
// UseLogger uses a specified Logger to output package logging info.
func UseLogger(logger btclog.Logger) {
log = logger
}
+628
View File
@@ -0,0 +1,628 @@
package migration4
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/tlv"
)
var (
// cChanDetailsBkt is a top-level bucket storing:
// channel-id => cChannelSummary -> encoded ClientChanSummary.
// => cChanDBID -> db-assigned-id
cChanDetailsBkt = []byte("client-channel-detail-bucket")
// cChanDBID is a key used in the cChanDetailsBkt to store the
// db-assigned-id of a channel.
cChanDBID = []byte("client-channel-db-id")
// cSessionBkt is a top-level bucket storing:
// session-id => cSessionBody -> encoded ClientSessionBody
// => cSessionCommits => seqnum -> encoded CommittedUpdate
// => cSessionAcks => seqnum -> encoded BackupID
cSessionBkt = []byte("client-session-bucket")
// cSessionAcks is a sub-bucket of cSessionBkt storing:
// seqnum -> encoded BackupID.
cSessionAcks = []byte("client-session-acks")
// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
// chan-id => start -> end
cSessionAckRangeIndex = []byte("client-session-ack-range-index")
// ErrUninitializedDB signals that top-level buckets for the database
// have not been initialized.
ErrUninitializedDB = errors.New("db not initialized")
// ErrClientSessionNotFound signals that the requested client session
// was not found in the database.
ErrClientSessionNotFound = errors.New("client session not found")
// ErrCorruptChanDetails signals that the clients channel detail's
// on-disk structure deviates from what is expected.
ErrCorruptChanDetails = errors.New("channel details corrupted")
// ErrChannelNotRegistered signals a channel has not yet been registered
// in the client database.
ErrChannelNotRegistered = errors.New("channel not registered")
// byteOrder is the default endianness used when serializing integers.
byteOrder = binary.BigEndian
// errExit is an error used to signal that the sessionIterator should
// exit.
errExit = errors.New("the exit condition has been met")
)
// DefaultSessionsPerTx is the default number of sessions that should be
// migrated per db transaction.
const DefaultSessionsPerTx = 5000
// MigrateAckedUpdates migrates the tower client DB. It takes the individual
// Acked Updates that are stored for each session and re-stores them using the
// RangeIndex representation.
func MigrateAckedUpdates(sessionsPerTx int) func(kvdb.Backend) error {
return func(db kvdb.Backend) error {
log.Infof("Migrating the tower client db to move all Acked " +
"Updates to the new Range Index representation.")
// Migrate the old acked-updates.
err := migrateAckedUpdates(db, sessionsPerTx)
if err != nil {
return fmt.Errorf("migration failed: %w", err)
}
log.Infof("Migrating old session acked updates finished, now " +
"checking the migration results...")
// Before we can safety delete the old buckets, we perform a
// check to make sure the sessions have been migrated as
// expected.
err = kvdb.View(db, validateMigration, func() {})
if err != nil {
return fmt.Errorf("validate migration failed: %w", err)
}
// Delete old acked updates.
err = kvdb.Update(db, deleteOldAckedUpdates, func() {})
if err != nil {
return fmt.Errorf("failed to delete old acked "+
"updates: %w", err)
}
return nil
}
}
// migrateAckedUpdates migrates the acked updates of each session in the
// wtclient db into the new RangeIndex form. This is done over multiple db
// transactions in order to prevent the migration from taking up too much RAM.
// The sessionsPerTx parameter can be used to set the maximum number of sessions
// that should be migrated per transaction.
func migrateAckedUpdates(db kvdb.Backend, sessionsPerTx int) error {
// Get migration progress stats.
total, migrated, err := logMigrationStats(db)
if err != nil {
return err
}
log.Infof("Total sessions=%d, migrated=%d", total, migrated)
// Exit early if the old session acked updates have already been
// migrated and deleted.
if total == 0 {
log.Info("Migration already finished!")
return nil
}
var (
finished bool
startKey []byte
)
for {
// Process the migration.
err = kvdb.Update(db, func(tx kvdb.RwTx) error {
startKey, finished, err = processMigration(
tx, startKey, sessionsPerTx,
)
return err
}, func() {})
if err != nil {
return err
}
if finished {
break
}
// Each time we finished the above process, we'd read the stats
// again to understand the current progress.
total, migrated, err = logMigrationStats(db)
if err != nil {
return err
}
// Calculate and log the progress if the progress is less than
// one hundred percent.
//nolint:gomnd
progress := float64(migrated) / float64(total) * 100
if progress >= 100 { //nolint:gomnd
break
}
log.Infof("Migration progress: %.3f%%, still have: %d",
progress, total-migrated)
}
return nil
}
func validateMigration(tx kvdb.RTx) error {
mainSessionsBkt := tx.ReadBucket(cSessionBkt)
if mainSessionsBkt == nil {
return ErrUninitializedDB
}
chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return ErrUninitializedDB
}
return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
// Get the bucket for this particular session.
sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)
if sessionBkt == nil {
return ErrClientSessionNotFound
}
// Get the bucket where any old acked updates would be stored.
oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)
// Get the bucket where any new acked updates would be stored.
newAcksBucket := sessionBkt.NestedReadBucket(
cSessionAckRangeIndex,
)
switch {
// If both the old and new acked updates buckets are nil, then
// we can safely skip this session.
case oldAcksBucket == nil && newAcksBucket == nil:
return nil
case oldAcksBucket == nil:
return fmt.Errorf("no old acks but do have new acks")
case newAcksBucket == nil:
return fmt.Errorf("no new acks but have old acks")
default:
}
// Collect acked ranges for this session.
ackedRanges := make(map[uint64]*RangeIndex)
err := newAcksBucket.ForEach(func(dbChanID, _ []byte) error {
rangeIndexBkt := newAcksBucket.NestedReadBucket(
dbChanID,
)
if rangeIndexBkt == nil {
return fmt.Errorf("no acked updates bucket "+
"found for channel %x", dbChanID)
}
// Read acked ranges from new bucket.
ri, err := readRangeIndex(rangeIndexBkt)
if err != nil {
return err
}
dbChanIDNum, err := readBigSize(dbChanID)
if err != nil {
return err
}
ackedRanges[dbChanIDNum] = ri
return nil
})
if err != nil {
return err
}
// Now we will iterate through each of the old acked updates and
// make sure that the update appears in the new bucket.
return oldAcksBucket.ForEach(func(_, v []byte) error {
var backupID BackupID
err := backupID.Decode(bytes.NewReader(v))
if err != nil {
return err
}
dbChanID, _, err := getDBChanID(
chanDetailsBkt, backupID.ChanID,
)
if err != nil {
return err
}
index, ok := ackedRanges[dbChanID]
if !ok {
return fmt.Errorf("no index found for this " +
"channel")
}
if !index.IsInIndex(backupID.CommitHeight) {
return fmt.Errorf("commit height not found " +
"in index")
}
return nil
})
})
}
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))
}
func deleteOldAckedUpdates(tx kvdb.RwTx) error {
mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)
if mainSessionsBkt == nil {
return ErrUninitializedDB
}
return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
// Get the bucket for this particular session.
sessionBkt := mainSessionsBkt.NestedReadWriteBucket(
sessID,
)
if sessionBkt == nil {
return ErrClientSessionNotFound
}
// Get the bucket where any old acked updates would be stored.
oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)
if oldAcksBucket == nil {
return nil
}
// Now that we have read everything that we need to from
// the cSessionAcks sub-bucket, we can delete it.
return sessionBkt.DeleteNestedBucket(cSessionAcks)
})
}
// processMigration uses the given transaction to perform a maximum of
// sessionsPerTx session migrations. If startKey is non-nil, it is used to
// determine the first session to start the migration at. The first return
// item is the key of the last session that was migrated successfully and the
// boolean is true if there are no more sessions left to migrate.
func processMigration(tx kvdb.RwTx, startKey []byte, sessionsPerTx int) ([]byte,
bool, error) {
chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
if chanDetailsBkt == nil {
return nil, false, ErrUninitializedDB
}
// sessionCount keeps track of the number of sessions that have been
// migrated under the current db transaction.
var sessionCount int
// migrateSessionCB is a callback function that calls migrateSession
// in order to migrate a single session. Upon success, the sessionCount
// is incremented and is then compared against sessionsPerTx to
// determine if we should continue migrating more sessions in this db
// transaction.
migrateSessionCB := func(sessionBkt kvdb.RwBucket) error {
err := migrateSession(chanDetailsBkt, sessionBkt)
if err != nil {
return err
}
sessionCount++
// If we have migrated sessionsPerTx sessions in this tx, then
// we return errExit in order to signal that this tx should be
// committed and the migration should be continued in a new
// transaction.
if sessionCount >= sessionsPerTx {
return errExit
}
return nil
}
// Starting at startKey, iterate over the sessions in the db and migrate
// them until either all are migrated or until the errExit signal is
// received.
lastKey, err := sessionIterator(tx, startKey, migrateSessionCB)
if err != nil && errors.Is(err, errExit) {
return lastKey, false, nil
} else if err != nil {
return nil, false, err
}
// The migration is complete.
return nil, true, nil
}
// migrateSession migrates a single session's acked-updates to the new
// RangeIndex form.
func migrateSession(chanDetailsBkt kvdb.RBucket,
sessionBkt kvdb.RwBucket) error {
// Get the existing cSessionAcks bucket. If there is no such bucket,
// then there are no acked-updates to migrate for this session.
sessionAcks := sessionBkt.NestedReadBucket(cSessionAcks)
if sessionAcks == nil {
return nil
}
// If there is already a new cSessionAckedRangeIndex bucket, then this
// session has already been migrated.
sessionAckRangesBkt := sessionBkt.NestedReadBucket(
cSessionAckRangeIndex,
)
if sessionAckRangesBkt != nil {
return nil
}
// Otherwise, we will iterate over each of the acked-updates, and we
// will construct a new RangeIndex for each channel.
m := make(map[ChannelID]*RangeIndex)
if err := sessionAcks.ForEach(func(_, v []byte) error {
var backupID BackupID
err := backupID.Decode(bytes.NewReader(v))
if err != nil {
return err
}
if _, ok := m[backupID.ChanID]; !ok {
index, err := NewRangeIndex(nil)
if err != nil {
return err
}
m[backupID.ChanID] = index
}
return m[backupID.ChanID].Add(backupID.CommitHeight, nil)
}); err != nil {
return err
}
// Create a new sub-bucket that will be used to store the new RangeIndex
// representation of the acked updates.
ackRangeBkt, err := sessionBkt.CreateBucket(cSessionAckRangeIndex)
if err != nil {
return err
}
// Iterate over each of the new range indexes that we will add for this
// session.
for chanID, rangeIndex := range m {
// Get db chanID.
chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
if chanDetails == nil {
return ErrCorruptChanDetails
}
// Create a sub-bucket for this channel using the db-assigned ID
// for the channel.
dbChanID := chanDetails.Get(cChanDBID)
chanAcksBkt, err := ackRangeBkt.CreateBucket(dbChanID)
if err != nil {
return err
}
// Iterate over the range pairs that we need to add to the DB.
for k, v := range rangeIndex.GetAllRanges() {
start, err := writeBigSize(k)
if err != nil {
return err
}
end, err := writeBigSize(v)
if err != nil {
return err
}
err = chanAcksBkt.Put(start, end)
if err != nil {
return err
}
}
}
return nil
}
// logMigrationStats reads the buckets to provide stats over current migration
// progress. The returned values are the numbers of total records and already
// migrated records.
func logMigrationStats(db kvdb.Backend) (uint64, uint64, error) {
var (
err error
total uint64
unmigrated uint64
)
err = kvdb.View(db, func(tx kvdb.RTx) error {
total, unmigrated, err = getMigrationStats(tx)
return err
}, func() {})
log.Debugf("Total sessions=%d, unmigrated=%d", total, unmigrated)
return total, total - unmigrated, err
}
// getMigrationStats iterates over all sessions. It counts the total number of
// sessions as well as the total number of unmigrated sessions.
func getMigrationStats(tx kvdb.RTx) (uint64, uint64, error) {
var (
total uint64
unmigrated uint64
)
// Get sessions bucket.
mainSessionsBkt := tx.ReadBucket(cSessionBkt)
if mainSessionsBkt == nil {
return 0, 0, ErrUninitializedDB
}
// Iterate over each session ID in the bucket.
err := mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
// Get the bucket for this particular session.
sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)
if sessionBkt == nil {
return ErrClientSessionNotFound
}
total++
// Get the cSessionAckRangeIndex bucket.
sessionAcksBkt := sessionBkt.NestedReadBucket(cSessionAcks)
// Get the cSessionAckRangeIndex bucket.
sessionAckRangesBkt := sessionBkt.NestedReadBucket(
cSessionAckRangeIndex,
)
// If both buckets do not exist, then this session is empty and
// does not need to be migrated.
if sessionAckRangesBkt == nil && sessionAcksBkt == nil {
return nil
}
// If the sessionAckRangesBkt is not nil, then the session has
// already been migrated.
if sessionAckRangesBkt != nil {
return nil
}
// Else the session has not yet been migrated.
unmigrated++
return nil
})
if err != nil {
return 0, 0, err
}
return total, unmigrated, nil
}
// getDBChanID returns the db-assigned channel ID for the given real channel ID.
// It returns both the uint64 and byte representation.
func getDBChanID(chanDetailsBkt kvdb.RBucket, chanID ChannelID) (uint64,
[]byte, error) {
chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
if chanDetails == nil {
return 0, nil, ErrChannelNotRegistered
}
idBytes := chanDetails.Get(cChanDBID)
if len(idBytes) == 0 {
return 0, nil, fmt.Errorf("no db-assigned ID found for "+
"channel ID %s", chanID)
}
id, err := readBigSize(idBytes)
if err != nil {
return 0, nil, err
}
return id, idBytes, nil
}
// callback defines a type that's used by the sessionIterator.
type callback func(bkt kvdb.RwBucket) error
// sessionIterator is a helper function that iterates over the main sessions
// bucket and performs the callback function on each individual session. If a
// seeker is specified, it will move the cursor to the given position otherwise
// it will start from the first item.
func sessionIterator(tx kvdb.RwTx, seeker []byte, cb callback) ([]byte, error) {
// Get sessions bucket.
mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)
if mainSessionsBkt == nil {
return nil, ErrUninitializedDB
}
c := mainSessionsBkt.ReadCursor()
k, _ := c.First()
// Move the cursor to the specified position if seeker is non-nil.
if seeker != nil {
k, _ = c.Seek(seeker)
}
// Start the iteration and exit on condition.
for k := k; k != nil; k, _ = c.Next() {
// Get the bucket for this particular session.
bkt := mainSessionsBkt.NestedReadWriteBucket(k)
if bkt == nil {
return nil, ErrClientSessionNotFound
}
// Call the callback function with the session's bucket.
if err := cb(bkt); err != nil {
// return k, err
lastIndex := make([]byte, len(k))
copy(lastIndex, k)
return lastIndex, err
}
}
return nil, nil
}
// writeBigSize will encode the given uint64 as a BigSize byte slice.
func writeBigSize(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
}
// readBigSize converts the given byte slice into a uint64 and assumes that the
// bytes slice is using BigSize encoding.
func readBigSize(b []byte) (uint64, error) {
r := bytes.NewReader(b)
i, err := tlv.ReadVarInt(r, &[8]byte{})
if err != nil {
return 0, err
}
return i, nil
}
@@ -0,0 +1,329 @@
package migration4
import (
"bytes"
"testing"
"github.com/btcsuite/btcwallet/walletdb"
"github.com/lightningnetwork/lnd/channeldb/migtest"
"github.com/lightningnetwork/lnd/kvdb"
)
var (
// details is the expected data of the channel details bucket. This
// bucket should not be changed during the migration, but it is used
// to find the db-assigned ID for each channel.
details = map[string]interface{}{
channelIDString(1): map[string]interface{}{
string(cChanDBID): uint64ToStr(10),
},
channelIDString(2): map[string]interface{}{
string(cChanDBID): uint64ToStr(20),
},
}
// preSessions is the expected data in the sessions bucket before the
// migration.
preSessions = map[string]interface{}{
sessionIDString("1"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 30,
}),
"2": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 31,
}),
"3": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 32,
}),
"4": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 34,
}),
"5": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 30,
}),
},
},
sessionIDString("2"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 33,
}),
},
},
sessionIDString("3"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 35,
}),
"2": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 36,
}),
"3": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 28,
}),
"4": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 29,
}),
},
},
}
// preMidStateDB is a possible state that the db could be in if the
// migration started but was interrupted before completing. In this
// state, some sessions still have the old cSessionAcks bucket along
// with the new cSessionAckRangeIndex. This is a valid pre-state and
// a migration on this state should succeed.
preMidStateDB = map[string]interface{}{
sessionIDString("1"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 30,
}),
"2": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 31,
}),
"3": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 32,
}),
"4": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 34,
}),
"5": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 30,
}),
},
string(cSessionAckRangeIndex): map[string]interface{}{
uint64ToStr(10): map[string]interface{}{
uint64ToStr(30): uint64ToStr(32),
uint64ToStr(34): uint64ToStr(34),
},
uint64ToStr(20): map[string]interface{}{
uint64ToStr(30): uint64ToStr(30),
},
},
},
sessionIDString("2"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 33,
}),
},
},
sessionIDString("3"): map[string]interface{}{
string(cSessionAcks): map[string]interface{}{
"1": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 35,
}),
"2": backupIDToString(&BackupID{
ChanID: intToChannelID(1),
CommitHeight: 36,
}),
"3": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 28,
}),
"4": backupIDToString(&BackupID{
ChanID: intToChannelID(2),
CommitHeight: 29,
}),
},
},
}
// preFailCorruptDB should fail the migration due to no session data
// being found for a given session ID.
preFailCorruptDB = map[string]interface{}{
sessionIDString("2"): "",
}
// postSessions is the expected data in the sessions bucket after the
// migration.
postSessions = map[string]interface{}{
sessionIDString("1"): map[string]interface{}{
string(cSessionAckRangeIndex): map[string]interface{}{
uint64ToStr(10): map[string]interface{}{
uint64ToStr(30): uint64ToStr(32),
uint64ToStr(34): uint64ToStr(34),
},
uint64ToStr(20): map[string]interface{}{
uint64ToStr(30): uint64ToStr(30),
},
},
},
sessionIDString("2"): map[string]interface{}{
string(cSessionAckRangeIndex): map[string]interface{}{
uint64ToStr(10): map[string]interface{}{
uint64ToStr(33): uint64ToStr(33),
},
},
},
sessionIDString("3"): map[string]interface{}{
string(cSessionAckRangeIndex): map[string]interface{}{
uint64ToStr(10): map[string]interface{}{
uint64ToStr(35): uint64ToStr(36),
},
uint64ToStr(20): map[string]interface{}{
uint64ToStr(28): uint64ToStr(29),
},
},
},
}
)
// TestMigrateAckedUpdates tests that the MigrateAckedUpdates function correctly
// migrates the existing AckedUpdates bucket for each session to the new
// RangeIndex representation.
func TestMigrateAckedUpdates(t *testing.T) {
t.Parallel()
tests := []struct {
name string
shouldFail bool
pre map[string]interface{}
post map[string]interface{}
}{
{
name: "migration ok",
shouldFail: false,
pre: preSessions,
post: postSessions,
},
{
name: "migration ok after re-starting",
shouldFail: false,
pre: preMidStateDB,
post: postSessions,
},
{
name: "fail due to corrupt db",
shouldFail: true,
pre: preFailCorruptDB,
},
{
name: "no sessions details",
shouldFail: false,
pre: nil,
post: nil,
},
}
for _, test := range tests {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
before := before(test.pre)
// After the migration, we should have an untouched
// summary bucket and a new index bucket.
after := after(test.shouldFail, test.pre, test.post)
migtest.ApplyMigrationWithDb(
t, before, after, MigrateAckedUpdates(2),
test.shouldFail,
)
})
}
}
// before returns a call-back function that can be used to set up a db's
// cChanDetailsBkt along with the cSessionBkt using the passed preMigDB
// structure.
func before(preMigDB map[string]interface{}) func(backend kvdb.Backend) error {
return func(db kvdb.Backend) error {
return db.Update(func(tx walletdb.ReadWriteTx) error {
err := migtest.RestoreDB(
tx, cChanDetailsBkt, details,
)
if err != nil {
return err
}
return migtest.RestoreDB(
tx, cSessionBkt, preMigDB,
)
}, func() {})
}
}
// after returns a call-back function that can be used to verify the state of
// a db post migration.
func after(shouldFail bool, preMigDB,
postMigDB map[string]interface{}) func(backend kvdb.Backend) error {
return func(db kvdb.Backend) error {
return db.Update(func(tx walletdb.ReadWriteTx) error {
// The channel details bucket should remain untouched.
err := migtest.VerifyDB(tx, cChanDetailsBkt, details)
if err != nil {
return err
}
// If the migration fails, the sessions bucket should be
// untouched.
if shouldFail {
if err := migtest.VerifyDB(
tx, cSessionBkt, preMigDB,
); err != nil {
return err
}
return nil
}
return migtest.VerifyDB(tx, cSessionBkt, postMigDB)
}, func() {})
}
}
func sessionIDString(id string) string {
var sessID SessionID
copy(sessID[:], id)
return sessID.String()
}
func intToChannelID(id uint64) ChannelID {
var chanID ChannelID
byteOrder.PutUint64(chanID[:], id)
return chanID
}
func channelIDString(id uint64) string {
var chanID ChannelID
byteOrder.PutUint64(chanID[:], id)
return string(chanID[:])
}
func uint64ToStr(id uint64) string {
b, err := writeBigSize(id)
if err != nil {
panic(err)
}
return string(b)
}
func backupIDToString(backup *BackupID) string {
var b bytes.Buffer
_ = backup.Encode(&b)
return b.String()
}
+129
View File
@@ -0,0 +1,129 @@
package migration4
import (
"encoding/binary"
"encoding/hex"
"fmt"
"io"
)
// SessionIDSize is 33-bytes; it is a serialized, compressed public key.
const SessionIDSize = 33
// SessionID is created from the remote public key of a client, and serves as a
// unique identifier and authentication for sending state updates.
type SessionID [SessionIDSize]byte
// String returns a hex encoding of the session id.
func (s SessionID) String() string {
return hex.EncodeToString(s[:])
}
// ChannelID is a series of 32-bytes that uniquely identifies all channels
// within the network. The ChannelID is computed using the outpoint of the
// funding transaction (the txid, and output index). Given a funding output the
// ChannelID can be calculated by XOR'ing the big-endian serialization of the
// txid and the big-endian serialization of the output index, truncated to
// 2 bytes.
type ChannelID [32]byte
// String returns the string representation of the ChannelID. This is just the
// hex string encoding of the ChannelID itself.
func (c ChannelID) String() string {
return hex.EncodeToString(c[:])
}
// BackupID identifies a particular revoked, remote commitment by channel id and
// commitment height.
type BackupID struct {
// ChanID is the channel id of the revoked commitment.
ChanID ChannelID
// CommitHeight is the commitment height of the revoked commitment.
CommitHeight uint64
}
// Encode writes the BackupID from the passed io.Writer.
func (b *BackupID) Encode(w io.Writer) error {
return WriteElements(w,
b.ChanID,
b.CommitHeight,
)
}
// Decode reads a BackupID from the passed io.Reader.
func (b *BackupID) Decode(r io.Reader) error {
return ReadElements(r,
&b.ChanID,
&b.CommitHeight,
)
}
// String returns a human-readable encoding of a BackupID.
func (b BackupID) String() string {
return fmt.Sprintf("backup(%v, %d)", b.ChanID, b.CommitHeight)
}
// WriteElements serializes a variadic list of elements into the given
// io.Writer.
func WriteElements(w io.Writer, elements ...interface{}) error {
for _, element := range elements {
if err := WriteElement(w, element); err != nil {
return err
}
}
return nil
}
// ReadElements deserializes the provided io.Reader into a variadic list of
// target elements.
func ReadElements(r io.Reader, elements ...interface{}) error {
for _, element := range elements {
if err := ReadElement(r, element); err != nil {
return err
}
}
return nil
}
// WriteElement serializes a single element into the provided io.Writer.
func WriteElement(w io.Writer, element interface{}) error {
switch e := element.(type) {
case ChannelID:
if _, err := w.Write(e[:]); err != nil {
return err
}
case uint64:
if err := binary.Write(w, byteOrder, e); err != nil {
return err
}
default:
return fmt.Errorf("unexpected type")
}
return nil
}
// ReadElement deserializes a single element from the provided io.Reader.
func ReadElement(r io.Reader, element interface{}) error {
switch e := element.(type) {
case *ChannelID:
if _, err := io.ReadFull(r, e[:]); err != nil {
return err
}
case *uint64:
if err := binary.Read(r, byteOrder, e); err != nil {
return err
}
default:
return fmt.Errorf("unexpected type")
}
return nil
}
+14
View File
@@ -0,0 +1,14 @@
package migration4
import (
"github.com/btcsuite/btclog"
)
// log is a logger that is initialized as disabled. This means the package will
// not perform any logging by default until a logger is set.
var log = btclog.Disabled
// UseLogger uses a specified Logger to output package logging info.
func UseLogger(logger btclog.Logger) {
log = logger
}
+619
View File
@@ -0,0 +1,619 @@
package migration4
import (
"fmt"
"sync"
)
// rangeItem represents the start and end values of a range.
type rangeItem struct {
start uint64
end uint64
}
// RangeIndexOption describes the signature of a functional option that can be
// used to modify the behaviour of a RangeIndex.
type RangeIndexOption func(*RangeIndex)
// WithSerializeUint64Fn is a functional option that can be used to set the
// function to be used to do the serialization of a uint64 into a byte slice.
func WithSerializeUint64Fn(fn func(uint64) ([]byte, error)) RangeIndexOption {
return func(index *RangeIndex) {
index.serializeUint64 = fn
}
}
// RangeIndex can be used to keep track of which numbers have been added to a
// set. It does so by keeping track of a sorted list of rangeItems. Each
// rangeItem has a start and end value of a range where all values in-between
// have been added to the set. It works well in situations where it is expected
// numbers in the set are not sparse.
type RangeIndex struct {
// set is a sorted list of rangeItem.
set []rangeItem
// mu is used to ensure safe access to set.
mu sync.Mutex
// serializeUint64 is the function that can be used to convert a uint64
// to a byte slice.
serializeUint64 func(uint64) ([]byte, error)
}
// NewRangeIndex constructs a new RangeIndex. An initial set of ranges may be
// passed to the function in the form of a map.
func NewRangeIndex(ranges map[uint64]uint64,
opts ...RangeIndexOption) (*RangeIndex, error) {
index := &RangeIndex{
serializeUint64: defaultSerializeUint64,
set: make([]rangeItem, 0),
}
// Apply any functional options.
for _, o := range opts {
o(index)
}
for s, e := range ranges {
if err := index.addRange(s, e); err != nil {
return nil, err
}
}
return index, nil
}
// addRange can be used to add an entire new range to the set. This method
// should only ever be called by NewRangeIndex to initialise the in-memory
// structure and so the RangeIndex mutex is not held during this method.
func (a *RangeIndex) addRange(start, end uint64) error {
// Check that the given range is valid.
if start > end {
return fmt.Errorf("invalid range. Start height %d is larger "+
"than end height %d", start, end)
}
// min is a helper closure that will return the minimum of two uint64s.
min := func(a, b uint64) uint64 {
if a < b {
return a
}
return b
}
// max is a helper closure that will return the maximum of two uint64s.
max := func(a, b uint64) uint64 {
if a > b {
return a
}
return b
}
// Collect the ranges that fall before and after the new range along
// with the start and end values of the new range.
var before, after []rangeItem
for _, x := range a.set {
// If the new start value can't extend the current ranges end
// value, then the two cannot be merged. The range is added to
// the group of ranges that fall before the new range.
if x.end+1 < start {
before = append(before, x)
continue
}
// If the current ranges start value does not follow on directly
// from the new end value, then the two cannot be merged. The
// range is added to the group of ranges that fall after the new
// range.
if end+1 < x.start {
after = append(after, x)
continue
}
// Otherwise, there is an overlap and so the two can be merged.
start = min(start, x.start)
end = max(end, x.end)
}
// Re-construct the range index set.
a.set = append(append(before, rangeItem{
start: start,
end: end,
}), after...)
return nil
}
// IsInIndex returns true if the given number is in the range set.
func (a *RangeIndex) IsInIndex(n uint64) bool {
a.mu.Lock()
defer a.mu.Unlock()
_, isCovered := a.lowerBoundIndex(n)
return isCovered
}
// NumInSet returns the number of items covered by the range set.
func (a *RangeIndex) NumInSet() uint64 {
a.mu.Lock()
defer a.mu.Unlock()
var numItems uint64
for _, r := range a.set {
numItems += r.end - r.start + 1
}
return numItems
}
// MaxHeight returns the highest number covered in the range.
func (a *RangeIndex) MaxHeight() uint64 {
a.mu.Lock()
defer a.mu.Unlock()
if len(a.set) == 0 {
return 0
}
return a.set[len(a.set)-1].end
}
// GetAllRanges returns a copy of the range set in the form of a map.
func (a *RangeIndex) GetAllRanges() map[uint64]uint64 {
a.mu.Lock()
defer a.mu.Unlock()
cp := make(map[uint64]uint64, len(a.set))
for _, item := range a.set {
cp[item.start] = item.end
}
return cp
}
// lowerBoundIndex returns the index of the RangeIndex that is most appropriate
// for the new value, n. In other words, it returns the index of the rangeItem
// set of the range where the start value is the highest start value in the set
// that is still lower than or equal to the given number, n. The returned
// boolean is true if the given number is already covered in the RangeIndex.
// A returned index of -1 indicates that no lower bound range exists in the set.
// Since the most likely case is that the new number will just extend the
// highest range, a check is first done to see if this is the case which will
// make the methods' computational complexity O(1). Otherwise, a binary search
// is done which brings the computational complexity to O(log N).
func (a *RangeIndex) lowerBoundIndex(n uint64) (int, bool) {
// If the set is empty, then there is no such index and the value
// definitely is not in the set.
if len(a.set) == 0 {
return -1, false
}
// In most cases, the last index item will be the one we want. So just
// do a quick check on that index first to avoid doing the binary
// search.
lastIndex := len(a.set) - 1
lastRange := a.set[lastIndex]
if lastRange.start <= n {
return lastIndex, lastRange.end >= n
}
// Otherwise, do a binary search to find the index of interest.
var (
low = 0
high = len(a.set) - 1
rangeIndex = -1
)
for {
mid := (low + high) / 2 //nolint: gomnd
currentRange := a.set[mid]
switch {
case currentRange.start > n:
// If the start of the range is greater than n, we can
// completely cut out that entire part of the array.
high = mid
case currentRange.start < n:
// If the range already includes the given height, we
// can stop searching now.
if currentRange.end >= n {
return mid, true
}
// If the start of the range is smaller than n, we can
// store this as the new best index to return.
rangeIndex = mid
// If low and mid are already equal, then increment low
// by 1. Exit if this means that low is now greater than
// high.
if low == mid {
low = mid + 1
if low > high {
return rangeIndex, false
}
} else {
low = mid
}
continue
default:
// If the height is equal to the start value of the
// current range that mid is pointing to, then the
// height is already covered.
return mid, true
}
// Exit if we have checked all the ranges.
if low == high {
break
}
}
return rangeIndex, false
}
// KVStore is an interface representing a key-value store.
type KVStore interface {
// Put saves the specified key/value pair to the store. Keys that do not
// already exist are added and keys that already exist are overwritten.
Put(key, value []byte) error
// Delete removes the specified key from the bucket. Deleting a key that
// does not exist does not return an error.
Delete(key []byte) error
}
// Add adds a single number to the range set. It first attempts to apply the
// necessary changes to the passed KV store and then only if this succeeds, will
// the changes be applied to the in-memory structure.
func (a *RangeIndex) Add(newHeight uint64, kv KVStore) error {
a.mu.Lock()
defer a.mu.Unlock()
// Compute the changes that will need to be applied to both the sorted
// rangeItem array representation and the key-value store representation
// of the range index.
arrayChanges, kvStoreChanges := a.getChanges(newHeight)
// First attempt to apply the KV store changes. Only if this succeeds
// will we apply the changes to our in-memory range index structure.
err := a.applyKVChanges(kv, kvStoreChanges)
if err != nil {
return err
}
// Since the DB changes were successful, we can now commit the
// changes to our in-memory representation of the range set.
a.applyArrayChanges(arrayChanges)
return nil
}
// applyKVChanges applies the given set of kvChanges to a KV store. It is
// assumed that a transaction is being held on the kv store so that if any
// of the actions of the function fails, the changes will be reverted.
func (a *RangeIndex) applyKVChanges(kv KVStore, changes *kvChanges) error {
// Exit early if there are no changes to apply.
if kv == nil || changes == nil {
return nil
}
// Check if any range pair needs to be deleted.
if changes.deleteKVKey != nil {
del, err := a.serializeUint64(*changes.deleteKVKey)
if err != nil {
return err
}
if err := kv.Delete(del); err != nil {
return err
}
}
start, err := a.serializeUint64(changes.key)
if err != nil {
return err
}
end, err := a.serializeUint64(changes.value)
if err != nil {
return err
}
return kv.Put(start, end)
}
// applyArrayChanges applies the given arrayChanges to the in-memory RangeIndex
// itself. This should only be done once the persisted kv store changes have
// already been applied.
func (a *RangeIndex) applyArrayChanges(changes *arrayChanges) {
if changes == nil {
return
}
if changes.indexToDelete != nil {
a.set = append(
a.set[:*changes.indexToDelete],
a.set[*changes.indexToDelete+1:]...,
)
}
if changes.newIndex != nil {
switch {
case *changes.newIndex == 0:
a.set = append([]rangeItem{{
start: changes.start,
end: changes.end,
}}, a.set...)
case *changes.newIndex == len(a.set):
a.set = append(a.set, rangeItem{
start: changes.start,
end: changes.end,
})
default:
a.set = append(
a.set[:*changes.newIndex+1],
a.set[*changes.newIndex:]...,
)
a.set[*changes.newIndex] = rangeItem{
start: changes.start,
end: changes.end,
}
}
return
}
if changes.indexToEdit != nil {
a.set[*changes.indexToEdit] = rangeItem{
start: changes.start,
end: changes.end,
}
}
}
// arrayChanges encompasses the diff to apply to the sorted rangeItem array
// representation of a range index. Such a diff will either include adding a
// new range or editing an existing range. If an existing range is edited, then
// the diff might also include deleting an index (this will be the case if the
// editing of the one range results in the merge of another range).
type arrayChanges struct {
start uint64
end uint64
// newIndex, if set, is the index of the in-memory range array where a
// new range, [start:end], should be added. newIndex should never be
// set at the same time as indexToEdit or indexToDelete.
newIndex *int
// indexToDelete, if set, is the index of the sorted rangeItem array
// that should be deleted. This should be applied before reading the
// index value of indexToEdit. This should not be set at the same time
// as newIndex.
indexToDelete *int
// indexToEdit is the index of the in-memory range array that should be
// edited. The range at this index will be changed to [start:end]. This
// should only be read after indexToDelete index has been deleted.
indexToEdit *int
}
// kvChanges encompasses the diff to apply to a KV-store representation of a
// range index. A kv-store diff for the addition of a single number to the range
// index will include either a brand new key-value pair or the altering of the
// value of an existing key. Optionally, the diff may also include the deletion
// of an existing key. A deletion will be required if the addition of the new
// number results in the merge of two ranges.
type kvChanges struct {
key uint64
value uint64
// deleteKVKey, if set, is the key of the kv store representation that
// should be deleted.
deleteKVKey *uint64
}
// getChanges will calculate and return the changes that need to be applied to
// both the sorted-rangeItem-array representation and the key-value store
// representation of the range index.
func (a *RangeIndex) getChanges(n uint64) (*arrayChanges, *kvChanges) {
// If the set is empty then a new range item is added.
if len(a.set) == 0 {
// For the array representation, a new range [n:n] is added to
// the first index of the array.
firstIndex := 0
ac := &arrayChanges{
newIndex: &firstIndex,
start: n,
end: n,
}
// For the KV representation, a new [n:n] pair is added.
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
}
// Find the index of the lower bound range to the new number.
indexOfRangeBelow, alreadyCovered := a.lowerBoundIndex(n)
switch {
// The new number is already covered by the range index. No changes are
// required.
case alreadyCovered:
return nil, nil
// No lower bound index exists.
case indexOfRangeBelow < 0:
// Check if the very first range can be merged into this new
// one.
if n+1 == a.set[0].start {
// If so, the two ranges can be merged and so the start
// value of the range is n and the end value is the end
// of the existing first range.
start := n
end := a.set[0].end
// For the array representation, we can just edit the
// first entry of the array
editIndex := 0
ac := &arrayChanges{
indexToEdit: &editIndex,
start: start,
end: end,
}
// For the KV store representation, we add a new kv pair
// and delete the range with the key equal to the start
// value of the range we are merging.
kvKeyToDelete := a.set[0].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &kvKeyToDelete,
}
return ac, kvc
}
// Otherwise, we add a new index.
// For the array representation, a new range [n:n] is added to
// the first index of the array.
newIndex := 0
ac := &arrayChanges{
newIndex: &newIndex,
start: n,
end: n,
}
// For the KV representation, a new [n:n] pair is added.
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
// A lower range does exist, and it can be extended to include this new
// number.
case a.set[indexOfRangeBelow].end+1 == n:
start := a.set[indexOfRangeBelow].start
end := n
indexToChange := indexOfRangeBelow
// If there are no intervals above this one or if there are, but
// they can't be merged into this one then we just need to edit
// this interval.
if indexOfRangeBelow == len(a.set)-1 ||
a.set[indexOfRangeBelow+1].start != n+1 {
// For the array representation, we just edit the index.
ac := &arrayChanges{
indexToEdit: &indexToChange,
start: start,
end: end,
}
// For the key-value representation, we just overwrite
// the end value at the existing start key.
kvc := &kvChanges{
key: start,
value: end,
}
return ac, kvc
}
// There is a range above this one that we need to merge into
// this one.
delIndex := indexOfRangeBelow + 1
end = a.set[delIndex].end
// For the array representation, we delete the range above this
// one and edit this range to include the end value of the range
// above.
ac := &arrayChanges{
indexToDelete: &delIndex,
indexToEdit: &indexToChange,
start: start,
end: end,
}
// For the kv representation, we tweak the end value of an
// existing key and delete the key of the range we are deleting.
deleteKey := a.set[delIndex].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &deleteKey,
}
return ac, kvc
// A lower range does exist, but it can't be extended to include this
// new number, and so we need to add a new range after the lower bound
// range.
default:
newIndex := indexOfRangeBelow + 1
// If there are no ranges above this new one or if there are,
// but they can't be merged into this new one, then we can just
// add the new one as is.
if newIndex == len(a.set) || a.set[newIndex].start != n+1 {
ac := &arrayChanges{
newIndex: &newIndex,
start: n,
end: n,
}
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
}
// Else, we merge the above index.
start := n
end := a.set[newIndex].end
toEdit := newIndex
// For the array representation, we edit the range above to
// include the new start value.
ac := &arrayChanges{
indexToEdit: &toEdit,
start: start,
end: end,
}
// For the kv representation, we insert the new start-end key
// value pair and delete the key using the old start value.
delKey := a.set[newIndex].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &delKey,
}
return ac, kvc
}
}
func defaultSerializeUint64(i uint64) ([]byte, error) {
var b [8]byte
byteOrder.PutUint64(b[:], i)
return b[:], nil
}
+619
View File
@@ -0,0 +1,619 @@
package wtdb
import (
"fmt"
"sync"
)
// rangeItem represents the start and end values of a range.
type rangeItem struct {
start uint64
end uint64
}
// RangeIndexOption describes the signature of a functional option that can be
// used to modify the behaviour of a RangeIndex.
type RangeIndexOption func(*RangeIndex)
// WithSerializeUint64Fn is a functional option that can be used to set the
// function to be used to do the serialization of a uint64 into a byte slice.
func WithSerializeUint64Fn(fn func(uint64) ([]byte, error)) RangeIndexOption {
return func(index *RangeIndex) {
index.serializeUint64 = fn
}
}
// RangeIndex can be used to keep track of which numbers have been added to a
// set. It does so by keeping track of a sorted list of rangeItems. Each
// rangeItem has a start and end value of a range where all values in-between
// have been added to the set. It works well in situations where it is expected
// numbers in the set are not sparse.
type RangeIndex struct {
// set is a sorted list of rangeItem.
set []rangeItem
// mu is used to ensure safe access to set.
mu sync.Mutex
// serializeUint64 is the function that can be used to convert a uint64
// to a byte slice.
serializeUint64 func(uint64) ([]byte, error)
}
// NewRangeIndex constructs a new RangeIndex. An initial set of ranges may be
// passed to the function in the form of a map.
func NewRangeIndex(ranges map[uint64]uint64,
opts ...RangeIndexOption) (*RangeIndex, error) {
index := &RangeIndex{
serializeUint64: defaultSerializeUint64,
set: make([]rangeItem, 0),
}
// Apply any functional options.
for _, o := range opts {
o(index)
}
for s, e := range ranges {
if err := index.addRange(s, e); err != nil {
return nil, err
}
}
return index, nil
}
// addRange can be used to add an entire new range to the set. This method
// should only ever be called by NewRangeIndex to initialise the in-memory
// structure and so the RangeIndex mutex is not held during this method.
func (a *RangeIndex) addRange(start, end uint64) error {
// Check that the given range is valid.
if start > end {
return fmt.Errorf("invalid range. Start height %d is larger "+
"than end height %d", start, end)
}
// min is a helper closure that will return the minimum of two uint64s.
min := func(a, b uint64) uint64 {
if a < b {
return a
}
return b
}
// max is a helper closure that will return the maximum of two uint64s.
max := func(a, b uint64) uint64 {
if a > b {
return a
}
return b
}
// Collect the ranges that fall before and after the new range along
// with the start and end values of the new range.
var before, after []rangeItem
for _, x := range a.set {
// If the new start value can't extend the current ranges end
// value, then the two cannot be merged. The range is added to
// the group of ranges that fall before the new range.
if x.end+1 < start {
before = append(before, x)
continue
}
// If the current ranges start value does not follow on directly
// from the new end value, then the two cannot be merged. The
// range is added to the group of ranges that fall after the new
// range.
if end+1 < x.start {
after = append(after, x)
continue
}
// Otherwise, there is an overlap and so the two can be merged.
start = min(start, x.start)
end = max(end, x.end)
}
// Re-construct the range index set.
a.set = append(append(before, rangeItem{
start: start,
end: end,
}), after...)
return nil
}
// IsInIndex returns true if the given number is in the range set.
func (a *RangeIndex) IsInIndex(n uint64) bool {
a.mu.Lock()
defer a.mu.Unlock()
_, isCovered := a.lowerBoundIndex(n)
return isCovered
}
// NumInSet returns the number of items covered by the range set.
func (a *RangeIndex) NumInSet() uint64 {
a.mu.Lock()
defer a.mu.Unlock()
var numItems uint64
for _, r := range a.set {
numItems += r.end - r.start + 1
}
return numItems
}
// MaxHeight returns the highest number covered in the range.
func (a *RangeIndex) MaxHeight() uint64 {
a.mu.Lock()
defer a.mu.Unlock()
if len(a.set) == 0 {
return 0
}
return a.set[len(a.set)-1].end
}
// GetAllRanges returns a copy of the range set in the form of a map.
func (a *RangeIndex) GetAllRanges() map[uint64]uint64 {
a.mu.Lock()
defer a.mu.Unlock()
cp := make(map[uint64]uint64, len(a.set))
for _, item := range a.set {
cp[item.start] = item.end
}
return cp
}
// lowerBoundIndex returns the index of the RangeIndex that is most appropriate
// for the new value, n. In other words, it returns the index of the rangeItem
// set of the range where the start value is the highest start value in the set
// that is still lower than or equal to the given number, n. The returned
// boolean is true if the given number is already covered in the RangeIndex.
// A returned index of -1 indicates that no lower bound range exists in the set.
// Since the most likely case is that the new number will just extend the
// highest range, a check is first done to see if this is the case which will
// make the methods' computational complexity O(1). Otherwise, a binary search
// is done which brings the computational complexity to O(log N).
func (a *RangeIndex) lowerBoundIndex(n uint64) (int, bool) {
// If the set is empty, then there is no such index and the value
// definitely is not in the set.
if len(a.set) == 0 {
return -1, false
}
// In most cases, the last index item will be the one we want. So just
// do a quick check on that index first to avoid doing the binary
// search.
lastIndex := len(a.set) - 1
lastRange := a.set[lastIndex]
if lastRange.start <= n {
return lastIndex, lastRange.end >= n
}
// Otherwise, do a binary search to find the index of interest.
var (
low = 0
high = len(a.set) - 1
rangeIndex = -1
)
for {
mid := (low + high) / 2 //nolint: gomnd
currentRange := a.set[mid]
switch {
case currentRange.start > n:
// If the start of the range is greater than n, we can
// completely cut out that entire part of the array.
high = mid
case currentRange.start < n:
// If the range already includes the given height, we
// can stop searching now.
if currentRange.end >= n {
return mid, true
}
// If the start of the range is smaller than n, we can
// store this as the new best index to return.
rangeIndex = mid
// If low and mid are already equal, then increment low
// by 1. Exit if this means that low is now greater than
// high.
if low == mid {
low = mid + 1
if low > high {
return rangeIndex, false
}
} else {
low = mid
}
continue
default:
// If the height is equal to the start value of the
// current range that mid is pointing to, then the
// height is already covered.
return mid, true
}
// Exit if we have checked all the ranges.
if low == high {
break
}
}
return rangeIndex, false
}
// KVStore is an interface representing a key-value store.
type KVStore interface {
// Put saves the specified key/value pair to the store. Keys that do not
// already exist are added and keys that already exist are overwritten.
Put(key, value []byte) error
// Delete removes the specified key from the bucket. Deleting a key that
// does not exist does not return an error.
Delete(key []byte) error
}
// Add adds a single number to the range set. It first attempts to apply the
// necessary changes to the passed KV store and then only if this succeeds, will
// the changes be applied to the in-memory structure.
func (a *RangeIndex) Add(newHeight uint64, kv KVStore) error {
a.mu.Lock()
defer a.mu.Unlock()
// Compute the changes that will need to be applied to both the sorted
// rangeItem array representation and the key-value store representation
// of the range index.
arrayChanges, kvStoreChanges := a.getChanges(newHeight)
// First attempt to apply the KV store changes. Only if this succeeds
// will we apply the changes to our in-memory range index structure.
err := a.applyKVChanges(kv, kvStoreChanges)
if err != nil {
return err
}
// Since the DB changes were successful, we can now commit the
// changes to our in-memory representation of the range set.
a.applyArrayChanges(arrayChanges)
return nil
}
// applyKVChanges applies the given set of kvChanges to a KV store. It is
// assumed that a transaction is being held on the kv store so that if any
// of the actions of the function fails, the changes will be reverted.
func (a *RangeIndex) applyKVChanges(kv KVStore, changes *kvChanges) error {
// Exit early if there are no changes to apply.
if kv == nil || changes == nil {
return nil
}
// Check if any range pair needs to be deleted.
if changes.deleteKVKey != nil {
del, err := a.serializeUint64(*changes.deleteKVKey)
if err != nil {
return err
}
if err := kv.Delete(del); err != nil {
return err
}
}
start, err := a.serializeUint64(changes.key)
if err != nil {
return err
}
end, err := a.serializeUint64(changes.value)
if err != nil {
return err
}
return kv.Put(start, end)
}
// applyArrayChanges applies the given arrayChanges to the in-memory RangeIndex
// itself. This should only be done once the persisted kv store changes have
// already been applied.
func (a *RangeIndex) applyArrayChanges(changes *arrayChanges) {
if changes == nil {
return
}
if changes.indexToDelete != nil {
a.set = append(
a.set[:*changes.indexToDelete],
a.set[*changes.indexToDelete+1:]...,
)
}
if changes.newIndex != nil {
switch {
case *changes.newIndex == 0:
a.set = append([]rangeItem{{
start: changes.start,
end: changes.end,
}}, a.set...)
case *changes.newIndex == len(a.set):
a.set = append(a.set, rangeItem{
start: changes.start,
end: changes.end,
})
default:
a.set = append(
a.set[:*changes.newIndex+1],
a.set[*changes.newIndex:]...,
)
a.set[*changes.newIndex] = rangeItem{
start: changes.start,
end: changes.end,
}
}
return
}
if changes.indexToEdit != nil {
a.set[*changes.indexToEdit] = rangeItem{
start: changes.start,
end: changes.end,
}
}
}
// arrayChanges encompasses the diff to apply to the sorted rangeItem array
// representation of a range index. Such a diff will either include adding a
// new range or editing an existing range. If an existing range is edited, then
// the diff might also include deleting an index (this will be the case if the
// editing of the one range results in the merge of another range).
type arrayChanges struct {
start uint64
end uint64
// newIndex, if set, is the index of the in-memory range array where a
// new range, [start:end], should be added. newIndex should never be
// set at the same time as indexToEdit or indexToDelete.
newIndex *int
// indexToDelete, if set, is the index of the sorted rangeItem array
// that should be deleted. This should be applied before reading the
// index value of indexToEdit. This should not be set at the same time
// as newIndex.
indexToDelete *int
// indexToEdit is the index of the in-memory range array that should be
// edited. The range at this index will be changed to [start:end]. This
// should only be read after indexToDelete index has been deleted.
indexToEdit *int
}
// kvChanges encompasses the diff to apply to a KV-store representation of a
// range index. A kv-store diff for the addition of a single number to the range
// index will include either a brand new key-value pair or the altering of the
// value of an existing key. Optionally, the diff may also include the deletion
// of an existing key. A deletion will be required if the addition of the new
// number results in the merge of two ranges.
type kvChanges struct {
key uint64
value uint64
// deleteKVKey, if set, is the key of the kv store representation that
// should be deleted.
deleteKVKey *uint64
}
// getChanges will calculate and return the changes that need to be applied to
// both the sorted-rangeItem-array representation and the key-value store
// representation of the range index.
func (a *RangeIndex) getChanges(n uint64) (*arrayChanges, *kvChanges) {
// If the set is empty then a new range item is added.
if len(a.set) == 0 {
// For the array representation, a new range [n:n] is added to
// the first index of the array.
firstIndex := 0
ac := &arrayChanges{
newIndex: &firstIndex,
start: n,
end: n,
}
// For the KV representation, a new [n:n] pair is added.
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
}
// Find the index of the lower bound range to the new number.
indexOfRangeBelow, alreadyCovered := a.lowerBoundIndex(n)
switch {
// The new number is already covered by the range index. No changes are
// required.
case alreadyCovered:
return nil, nil
// No lower bound index exists.
case indexOfRangeBelow < 0:
// Check if the very first range can be merged into this new
// one.
if n+1 == a.set[0].start {
// If so, the two ranges can be merged and so the start
// value of the range is n and the end value is the end
// of the existing first range.
start := n
end := a.set[0].end
// For the array representation, we can just edit the
// first entry of the array
editIndex := 0
ac := &arrayChanges{
indexToEdit: &editIndex,
start: start,
end: end,
}
// For the KV store representation, we add a new kv pair
// and delete the range with the key equal to the start
// value of the range we are merging.
kvKeyToDelete := a.set[0].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &kvKeyToDelete,
}
return ac, kvc
}
// Otherwise, we add a new index.
// For the array representation, a new range [n:n] is added to
// the first index of the array.
newIndex := 0
ac := &arrayChanges{
newIndex: &newIndex,
start: n,
end: n,
}
// For the KV representation, a new [n:n] pair is added.
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
// A lower range does exist, and it can be extended to include this new
// number.
case a.set[indexOfRangeBelow].end+1 == n:
start := a.set[indexOfRangeBelow].start
end := n
indexToChange := indexOfRangeBelow
// If there are no intervals above this one or if there are, but
// they can't be merged into this one then we just need to edit
// this interval.
if indexOfRangeBelow == len(a.set)-1 ||
a.set[indexOfRangeBelow+1].start != n+1 {
// For the array representation, we just edit the index.
ac := &arrayChanges{
indexToEdit: &indexToChange,
start: start,
end: end,
}
// For the key-value representation, we just overwrite
// the end value at the existing start key.
kvc := &kvChanges{
key: start,
value: end,
}
return ac, kvc
}
// There is a range above this one that we need to merge into
// this one.
delIndex := indexOfRangeBelow + 1
end = a.set[delIndex].end
// For the array representation, we delete the range above this
// one and edit this range to include the end value of the range
// above.
ac := &arrayChanges{
indexToDelete: &delIndex,
indexToEdit: &indexToChange,
start: start,
end: end,
}
// For the kv representation, we tweak the end value of an
// existing key and delete the key of the range we are deleting.
deleteKey := a.set[delIndex].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &deleteKey,
}
return ac, kvc
// A lower range does exist, but it can't be extended to include this
// new number, and so we need to add a new range after the lower bound
// range.
default:
newIndex := indexOfRangeBelow + 1
// If there are no ranges above this new one or if there are,
// but they can't be merged into this new one, then we can just
// add the new one as is.
if newIndex == len(a.set) || a.set[newIndex].start != n+1 {
ac := &arrayChanges{
newIndex: &newIndex,
start: n,
end: n,
}
kvc := &kvChanges{
key: n,
value: n,
}
return ac, kvc
}
// Else, we merge the above index.
start := n
end := a.set[newIndex].end
toEdit := newIndex
// For the array representation, we edit the range above to
// include the new start value.
ac := &arrayChanges{
indexToEdit: &toEdit,
start: start,
end: end,
}
// For the kv representation, we insert the new start-end key
// value pair and delete the key using the old start value.
delKey := a.set[newIndex].start
kvc := &kvChanges{
key: start,
value: end,
deleteKVKey: &delKey,
}
return ac, kvc
}
}
func defaultSerializeUint64(i uint64) ([]byte, error) {
var b [8]byte
byteOrder.PutUint64(b[:], i)
return b[:], nil
}
+349
View File
@@ -0,0 +1,349 @@
package wtdb
import (
"fmt"
"math/rand"
"testing"
"time"
"github.com/stretchr/testify/require"
)
// TestRangeIndex tests that the RangeIndex works as expected.
func TestRangeIndex(t *testing.T) {
t.Parallel()
assertRanges := func(index *RangeIndex, kvStore *mockKVStore,
v map[uint64]uint64) {
require.EqualValues(t, v, index.GetAllRanges())
require.EqualValues(t, v, kvStore.kv)
}
t.Run("test zero value height", func(t *testing.T) {
t.Parallel()
kvStore := newMockKVStore(nil)
index, err := NewRangeIndex(nil)
require.NoError(t, err)
// Since zero values are tricky, assert that the empty index
// does not include zero.
require.False(t, index.IsInIndex(0))
// Now add zero to the index.
_ = index.Add(0, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{0: 0})
require.True(t, index.IsInIndex(0))
})
t.Run("add duplicates", func(t *testing.T) {
t.Parallel()
kvStore := newMockKVStore(nil)
index, err := NewRangeIndex(nil)
require.NoError(t, err)
require.False(t, index.IsInIndex(1))
// Add 1 to the range.
err = index.Add(1, kvStore)
require.NoError(t, err)
assertRanges(index, kvStore, map[uint64]uint64{1: 1})
require.EqualValues(t, 1, index.MaxHeight())
require.True(t, index.IsInIndex(1))
// Add 1 again and assert that nothing has changed.
err = index.Add(1, kvStore)
require.NoError(t, err)
assertRanges(index, kvStore, map[uint64]uint64{1: 1})
require.EqualValues(t, 1, index.MaxHeight())
require.True(t, index.IsInIndex(1))
})
t.Run("extend an existing range", func(t *testing.T) {
t.Parallel()
kvStore := newMockKVStore(nil)
index, err := NewRangeIndex(nil)
require.NoError(t, err)
assertRanges(index, kvStore, map[uint64]uint64{})
// Add 2.
_ = index.Add(2, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
2: 2,
})
// Add 3, 4 and 5 and assert that these just extend the existing
// range by incrementing its end value.
_ = index.Add(3, kvStore)
_ = index.Add(4, kvStore)
_ = index.Add(5, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
2: 5,
})
// Now add 1 and 0 and assert that these just extend the
// existing range by decrementing its start value.
_ = index.Add(1, kvStore)
_ = index.Add(0, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
0: 5,
})
// Assert various other properties of the current range.
require.True(t, index.IsInIndex(3))
require.EqualValues(t, 5, index.MaxHeight())
})
t.Run("add new ranges above and below", func(t *testing.T) {
t.Parallel()
// Initialise the index with an initial range.
initialState := map[uint64]uint64{
4: 10,
}
kvStore := newMockKVStore(initialState)
index, err := NewRangeIndex(initialState)
require.NoError(t, err)
// Add 2 and 12. This should create two new ranges in the index.
_ = index.Add(12, kvStore)
_ = index.Add(2, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
2: 2,
4: 10,
12: 12,
})
// Assert various other properties of the current range.
require.EqualValues(t, 12, index.MaxHeight())
require.False(t, index.IsInIndex(3))
require.False(t, index.IsInIndex(11))
require.True(t, index.IsInIndex(2))
require.True(t, index.IsInIndex(5))
require.True(t, index.IsInIndex(12))
})
t.Run("merging two ranges", func(t *testing.T) {
t.Parallel()
// Initialise the index with an initial set of ranges.
initialState := map[uint64]uint64{
2: 2,
4: 10,
12: 12,
}
kvStore := newMockKVStore(initialState)
index, err := NewRangeIndex(initialState)
require.NoError(t, err)
// Adding 3 should merge the first and second ranges.
_ = index.Add(3, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
2: 10,
12: 12,
})
// Adding 11 should merge the first and second ranges.
_ = index.Add(11, kvStore)
assertRanges(index, kvStore, map[uint64]uint64{
2: 12,
})
// Assert various other properties of the current range.
require.EqualValues(t, 12, index.MaxHeight())
require.True(t, index.IsInIndex(2))
require.True(t, index.IsInIndex(5))
require.True(t, index.IsInIndex(12))
require.False(t, index.IsInIndex(1))
})
t.Run("failure applying KV store updates", func(t *testing.T) {
t.Parallel()
kvStore := newMockKVStore(nil)
index, err := NewRangeIndex(nil)
require.NoError(t, err)
assertRanges(index, kvStore, map[uint64]uint64{})
// Ensure that the kv store will return an error when its
// methods are called.
kvStore.setError(fmt.Errorf("db error"))
// Now attempt to add a new item to the range.
err = index.Add(20, kvStore)
require.Error(t, err)
// Assert that the update failed for both the kv store and the
// array store.
assertRanges(index, kvStore, map[uint64]uint64{})
// Now let the kv store again not return an error.
kvStore.setError(nil)
// Again attempt to add a new item to the range.
err = index.Add(20, kvStore)
require.NoError(t, err)
// It should now succeed.
assertRanges(index, kvStore, map[uint64]uint64{
20: 20,
})
})
t.Run("initialising with different ranges", func(t *testing.T) {
t.Parallel()
tests := []struct {
name string
input map[uint64]uint64
expectErr bool
expectedIndex map[uint64]uint64
}{
{
name: "invalid ranges",
input: map[uint64]uint64{
1: 5,
6: 4,
},
expectErr: true,
},
{
name: "non-overlapping ranges",
input: map[uint64]uint64{
1: 2,
4: 6,
8: 20,
},
expectedIndex: map[uint64]uint64{
1: 2,
4: 6,
8: 20,
},
},
{
name: "merge-able ranges",
input: map[uint64]uint64{
1: 2,
3: 6,
7: 20,
},
expectedIndex: map[uint64]uint64{
1: 20,
},
},
{
name: "overlapping ranges",
input: map[uint64]uint64{
1: 4,
3: 7,
6: 20,
},
expectedIndex: map[uint64]uint64{
1: 20,
},
},
}
for _, test := range tests {
index, err := NewRangeIndex(test.input)
if test.expectErr {
require.Error(t, err)
continue
}
require.NoError(t, err)
require.EqualValues(
t, test.expectedIndex, index.GetAllRanges(),
)
}
})
t.Run("test large number of random inserts", func(t *testing.T) {
t.Parallel()
size := 10000
// Construct an array with values from 0 to size.
arr := make([]uint64, size)
for i := 0; i < size; i++ {
arr[i] = uint64(i)
}
// Shuffle the array so that the values are not added in order.
rand.Seed(time.Now().UnixNano())
rand.Shuffle(len(arr), func(i, j int) {
arr[i], arr[j] = arr[j], arr[i]
})
kvStore := newMockKVStore(nil)
index, err := NewRangeIndex(nil)
require.NoError(t, err)
// Now add each item in the array to the index.
for _, n := range arr {
_ = index.Add(n, kvStore)
}
// Assert that in the end, there is only a single range.
assertRanges(index, kvStore, map[uint64]uint64{
0: uint64(size - 1),
})
require.EqualValues(t, uint64(size-1), index.MaxHeight())
})
}
type mockKVStore struct {
kv map[uint64]uint64
err error
}
func newMockKVStore(initialRanges map[uint64]uint64) *mockKVStore {
if initialRanges != nil {
return &mockKVStore{
kv: initialRanges,
}
}
return &mockKVStore{
kv: make(map[uint64]uint64),
}
}
func (m *mockKVStore) setError(err error) {
m.err = err
}
func (m *mockKVStore) Put(key, value []byte) error {
if m.err != nil {
return m.err
}
k := byteOrder.Uint64(key)
v := byteOrder.Uint64(value)
m.kv[k] = v
return nil
}
func (m *mockKVStore) Delete(key []byte) error {
if m.err != nil {
return m.err
}
k := byteOrder.Uint64(key)
delete(m.kv, k)
return nil
}
+76 -16
View File
@@ -1,20 +1,35 @@
package wtdb
import (
"fmt"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration1"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration2"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration3"
"github.com/lightningnetwork/lnd/watchtower/wtdb/migration4"
)
// 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
// txMigration 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. It uses an existing database transaction
// to do so.
type txMigration func(tx kvdb.RwTx) error
// dbMigration 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. If such a migration is defined, the
// migration is responsible for starting any database transactions. This
// migration type is useful in the case where it would be beneficial (in terms
// of RAM usage) to split the migration between multiple db transactions.
type dbMigration func(db kvdb.Backend) error
// version pairs a version number with the migration that would need to be
// applied from the prior version to upgrade.
type version struct {
migration migration
txMigration txMigration
dbMigration dbMigration
}
// towerDBVersions stores all versions and migrations of the tower database.
@@ -27,7 +42,18 @@ var towerDBVersions = []version{}
// migrations must be applied.
var clientDBVersions = []version{
{
migration: migration1.MigrateTowerToSessionIndex,
txMigration: migration1.MigrateTowerToSessionIndex,
},
{
txMigration: migration2.MigrateClientChannelDetails,
},
{
txMigration: migration3.MigrateChannelIDIndex,
},
{
dbMigration: migration4.MigrateAckedUpdates(
migration4.DefaultSessionsPerTx,
),
},
}
@@ -146,23 +172,57 @@ func syncVersions(db versionedDB, versions []version) error {
// Otherwise, apply any migrations in order to bring the database
// version up to the highest known version.
updates := getMigrations(versions, curVersion)
return kvdb.Update(db.bdb(), func(tx kvdb.RwTx) error {
for i, update := range updates {
if update.migration == nil {
continue
}
for i, update := range updates {
if update.dbMigration != nil && update.txMigration != nil {
return fmt.Errorf("cannot specify both a " +
"tx-migration and a db-migration for a " +
"single version")
}
version := curVersion + uint32(i) + 1
log.Infof("Applying migration #%d", version)
version := curVersion + uint32(i) + 1
log.Infof("Applying migration #%d", version)
err := update.migration(tx)
if update.dbMigration != nil {
err = update.dbMigration(db.bdb())
if err != nil {
log.Errorf("Unable to apply migration #%d: %v",
version, err)
return err
}
// Note that unlike a txMigration, here we update the
// db version in a transaction that is separate to the
// transaction in which the db migration took place.
// This means that the db migration function must be
// idempotent.
err = kvdb.Update(db.bdb(), func(tx kvdb.RwTx) error {
return putDBVersion(tx, version)
}, func() {})
if err != nil {
return err
}
continue
}
return putDBVersion(tx, latestVersion)
}, func() {})
if update.txMigration == nil {
continue
}
err = kvdb.Update(db.bdb(), func(tx kvdb.RwTx) error {
err := update.txMigration(tx)
if err != nil {
log.Errorf("Unable to apply migration #%d: %v",
version, err)
return err
}
return putDBVersion(tx, version)
}, func() {})
if err != nil {
return err
}
}
return nil
}
+134 -20
View File
@@ -1,6 +1,7 @@
package wtmock
import (
"encoding/binary"
"net"
"sync"
"sync/atomic"
@@ -11,6 +12,8 @@ import (
"github.com/lightningnetwork/lnd/watchtower/wtdb"
)
var byteOrder = binary.BigEndian
type towerPK [33]byte
type keyIndexKey struct {
@@ -18,18 +21,23 @@ type keyIndexKey struct {
blobType blob.Type
}
type rangeIndexArrayMap map[wtdb.SessionID]map[lnwire.ChannelID]*wtdb.RangeIndex
type rangeIndexKVStore map[wtdb.SessionID]map[lnwire.ChannelID]*mockKVStore
// ClientDB is a mock, in-memory database or testing the watchtower client
// behavior.
type ClientDB struct {
nextTowerID uint64 // to be used atomically
mu sync.Mutex
summaries map[lnwire.ChannelID]wtdb.ClientChanSummary
activeSessions map[wtdb.SessionID]wtdb.ClientSession
ackedUpdates map[wtdb.SessionID]map[uint16]wtdb.BackupID
committedUpdates map[wtdb.SessionID][]wtdb.CommittedUpdate
towerIndex map[towerPK]wtdb.TowerID
towers map[wtdb.TowerID]*wtdb.Tower
mu sync.Mutex
summaries map[lnwire.ChannelID]wtdb.ClientChanSummary
activeSessions map[wtdb.SessionID]wtdb.ClientSession
ackedUpdates rangeIndexArrayMap
persistedAckedUpdates rangeIndexKVStore
committedUpdates map[wtdb.SessionID][]wtdb.CommittedUpdate
towerIndex map[towerPK]wtdb.TowerID
towers map[wtdb.TowerID]*wtdb.Tower
nextIndex uint32
indexes map[keyIndexKey]uint32
@@ -39,14 +47,21 @@ type ClientDB struct {
// NewClientDB initializes a new mock ClientDB.
func NewClientDB() *ClientDB {
return &ClientDB{
summaries: make(map[lnwire.ChannelID]wtdb.ClientChanSummary),
activeSessions: make(map[wtdb.SessionID]wtdb.ClientSession),
ackedUpdates: make(map[wtdb.SessionID]map[uint16]wtdb.BackupID),
committedUpdates: make(map[wtdb.SessionID][]wtdb.CommittedUpdate),
towerIndex: make(map[towerPK]wtdb.TowerID),
towers: make(map[wtdb.TowerID]*wtdb.Tower),
indexes: make(map[keyIndexKey]uint32),
legacyIndexes: make(map[wtdb.TowerID]uint32),
summaries: make(
map[lnwire.ChannelID]wtdb.ClientChanSummary,
),
activeSessions: make(
map[wtdb.SessionID]wtdb.ClientSession,
),
ackedUpdates: make(rangeIndexArrayMap),
persistedAckedUpdates: make(rangeIndexKVStore),
committedUpdates: make(
map[wtdb.SessionID][]wtdb.CommittedUpdate,
),
towerIndex: make(map[towerPK]wtdb.TowerID),
towers: make(map[wtdb.TowerID]*wtdb.Tower),
indexes: make(map[keyIndexKey]uint32),
legacyIndexes: make(map[wtdb.TowerID]uint32),
}
}
@@ -233,9 +248,20 @@ func (m *ClientDB) listClientSessions(tower *wtdb.TowerID,
}
sessions[session.ID] = &session
if cfg.PerAckedUpdate != nil {
for seq, id := range m.ackedUpdates[session.ID] {
cfg.PerAckedUpdate(&session, seq, id)
if cfg.PerMaxHeight != nil {
for chanID, index := range m.ackedUpdates[session.ID] {
cfg.PerMaxHeight(
&session, chanID, index.MaxHeight(),
)
}
}
if cfg.PerNumAckedUpdates != nil {
for chanID, index := range m.ackedUpdates[session.ID] {
cfg.PerNumAckedUpdates(
&session, chanID,
uint16(index.NumInSet()),
)
}
}
@@ -266,6 +292,37 @@ func (m *ClientDB) FetchSessionCommittedUpdates(id *wtdb.SessionID) (
return updates, nil
}
// IsAcked returns true if the given backup has been backed up using the given
// session.
func (m *ClientDB) IsAcked(id *wtdb.SessionID, backupID *wtdb.BackupID) (bool,
error) {
m.mu.Lock()
defer m.mu.Unlock()
index, ok := m.ackedUpdates[*id][backupID.ChanID]
if !ok {
return false, nil
}
return index.IsInIndex(backupID.CommitHeight), nil
}
// NumAckedUpdates returns the number of backups that have been successfully
// backed up using the given session.
func (m *ClientDB) NumAckedUpdates(id *wtdb.SessionID) (uint64, error) {
m.mu.Lock()
defer m.mu.Unlock()
var numAcked uint64
for _, index := range m.ackedUpdates[*id] {
numAcked += index.NumInSet()
}
return numAcked, nil
}
// CreateClientSession records a newly negotiated client session in the set of
// active sessions. The session can be identified by its SessionID.
func (m *ClientDB) CreateClientSession(session *wtdb.ClientSession) error {
@@ -311,7 +368,10 @@ func (m *ClientDB) CreateClientSession(session *wtdb.ClientSession) error {
RewardPkScript: cloneBytes(session.RewardPkScript),
},
}
m.ackedUpdates[session.ID] = make(map[uint16]wtdb.BackupID)
m.ackedUpdates[session.ID] = make(map[lnwire.ChannelID]*wtdb.RangeIndex)
m.persistedAckedUpdates[session.ID] = make(
map[lnwire.ChannelID]*mockKVStore,
)
m.committedUpdates[session.ID] = make([]wtdb.CommittedUpdate, 0)
return nil
@@ -443,7 +503,25 @@ func (m *ClientDB) AckUpdate(id *wtdb.SessionID, seqNum,
updates[len(updates)-1] = wtdb.CommittedUpdate{}
m.committedUpdates[session.ID] = updates[:len(updates)-1]
m.ackedUpdates[*id][seqNum] = update.BackupID
chanID := update.BackupID.ChanID
if _, ok := m.ackedUpdates[*id][update.BackupID.ChanID]; !ok {
index, err := wtdb.NewRangeIndex(nil)
if err != nil {
return err
}
m.ackedUpdates[*id][chanID] = index
m.persistedAckedUpdates[*id][chanID] = newMockKVStore()
}
err := m.ackedUpdates[*id][chanID].Add(
update.BackupID.CommitHeight,
m.persistedAckedUpdates[*id][chanID],
)
if err != nil {
return err
}
session.TowerLastApplied = lastApplied
m.activeSessions[*id] = session
@@ -512,3 +590,39 @@ func copyTower(tower *wtdb.Tower) *wtdb.Tower {
return t
}
type mockKVStore struct {
kv map[uint64]uint64
err error
}
func newMockKVStore() *mockKVStore {
return &mockKVStore{
kv: make(map[uint64]uint64),
}
}
func (m *mockKVStore) Put(key, value []byte) error {
if m.err != nil {
return m.err
}
k := byteOrder.Uint64(key)
v := byteOrder.Uint64(value)
m.kv[k] = v
return nil
}
func (m *mockKVStore) Delete(key []byte) error {
if m.err != nil {
return m.err
}
k := byteOrder.Uint64(key)
delete(m.kv, k)
return nil
}