diff --git a/channeldb/migration30/migration_test.go b/channeldb/migration30/migration_test.go index c748ebac0..5ec91fdff 100644 --- a/channeldb/migration30/migration_test.go +++ b/channeldb/migration30/migration_test.go @@ -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, ) } diff --git a/channeldb/migtest/migtest.go b/channeldb/migtest/migtest.go index 65e36308f..dacca7dac 100644 --- a/channeldb/migtest/migtest.go +++ b/channeldb/migtest/migtest.go @@ -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. diff --git a/docs/release-notes/release-notes-0.16.0.md b/docs/release-notes/release-notes-0.16.0.md index 7fd1358ff..e06c78366 100644 --- a/docs/release-notes/release-notes-0.16.0.md +++ b/docs/release-notes/release-notes-0.16.0.md @@ -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 diff --git a/lnrpc/wtclientrpc/wtclient.go b/lnrpc/wtclientrpc/wtclient.go index 1d9d41849..6672c0c75 100644 --- a/lnrpc/wtclientrpc/wtclient.go +++ b/lnrpc/wtclientrpc/wtclient.go @@ -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 { diff --git a/watchtower/log.go b/watchtower/log.go index 8ce96131e..8e9062d95 100644 --- a/watchtower/log.go +++ b/watchtower/log.go @@ -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) } diff --git a/watchtower/wtclient/client.go b/watchtower/wtclient/client.go index 3aa84f28c..685170eab 100644 --- a/watchtower/wtclient/client.go +++ b/watchtower/wtclient/client.go @@ -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 { diff --git a/watchtower/wtclient/interface.go b/watchtower/wtclient/interface.go index ba6546328..63d718b60 100644 --- a/watchtower/wtclient/interface.go +++ b/watchtower/wtclient/interface.go @@ -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) diff --git a/watchtower/wtdb/client_db.go b/watchtower/wtdb/client_db.go index 26d4704d4..f8eda37bf 100644 --- a/watchtower/wtdb/client_db.go +++ b/watchtower/wtdb/client_db.go @@ -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 +} diff --git a/watchtower/wtdb/client_db_test.go b/watchtower/wtdb/client_db_test.go index f75a0c2bc..2cafc160b 100644 --- a/watchtower/wtdb/client_db_test.go +++ b/watchtower/wtdb/client_db_test.go @@ -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. diff --git a/watchtower/wtdb/codec_test.go b/watchtower/wtdb/codec_test.go index c2628b86a..228d2a0ec 100644 --- a/watchtower/wtdb/codec_test.go +++ b/watchtower/wtdb/codec_test.go @@ -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 diff --git a/watchtower/wtdb/log.go b/watchtower/wtdb/log.go index 6ddb6c35f..134cee4a7 100644 --- a/watchtower/wtdb/log.go +++ b/watchtower/wtdb/log.go @@ -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 diff --git a/watchtower/wtdb/migration2/client_db.go b/watchtower/wtdb/migration2/client_db.go new file mode 100644 index 000000000..ead2b9ea3 --- /dev/null +++ b/watchtower/wtdb/migration2/client_db.go @@ -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) +} diff --git a/watchtower/wtdb/migration2/client_db_test.go b/watchtower/wtdb/migration2/client_db_test.go new file mode 100644 index 000000000..c1436184f --- /dev/null +++ b/watchtower/wtdb/migration2/client_db_test.go @@ -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() +} diff --git a/watchtower/wtdb/migration2/codec.go b/watchtower/wtdb/migration2/codec.go new file mode 100644 index 000000000..e3107ba8b --- /dev/null +++ b/watchtower/wtdb/migration2/codec.go @@ -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[:]) +} diff --git a/watchtower/wtdb/migration2/log.go b/watchtower/wtdb/migration2/log.go new file mode 100644 index 000000000..fd64d2fcb --- /dev/null +++ b/watchtower/wtdb/migration2/log.go @@ -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 +} diff --git a/watchtower/wtdb/migration3/client_db.go b/watchtower/wtdb/migration3/client_db.go new file mode 100644 index 000000000..334a89abe --- /dev/null +++ b/watchtower/wtdb/migration3/client_db.go @@ -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) + }) +} diff --git a/watchtower/wtdb/migration3/client_db_test.go b/watchtower/wtdb/migration3/client_db_test.go new file mode 100644 index 000000000..ffe0ee31f --- /dev/null +++ b/watchtower/wtdb/migration3/client_db_test.go @@ -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() +} diff --git a/watchtower/wtdb/migration3/codec.go b/watchtower/wtdb/migration3/codec.go new file mode 100644 index 000000000..402c33ce9 --- /dev/null +++ b/watchtower/wtdb/migration3/codec.go @@ -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[:]) +} diff --git a/watchtower/wtdb/migration3/log.go b/watchtower/wtdb/migration3/log.go new file mode 100644 index 000000000..2f14ff5e7 --- /dev/null +++ b/watchtower/wtdb/migration3/log.go @@ -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 +} diff --git a/watchtower/wtdb/migration4/client_db.go b/watchtower/wtdb/migration4/client_db.go new file mode 100644 index 000000000..0dcb499c8 --- /dev/null +++ b/watchtower/wtdb/migration4/client_db.go @@ -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 +} diff --git a/watchtower/wtdb/migration4/client_db_test.go b/watchtower/wtdb/migration4/client_db_test.go new file mode 100644 index 000000000..2cdb7c4e9 --- /dev/null +++ b/watchtower/wtdb/migration4/client_db_test.go @@ -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() +} diff --git a/watchtower/wtdb/migration4/codec.go b/watchtower/wtdb/migration4/codec.go new file mode 100644 index 000000000..9205f6528 --- /dev/null +++ b/watchtower/wtdb/migration4/codec.go @@ -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 +} diff --git a/watchtower/wtdb/migration4/log.go b/watchtower/wtdb/migration4/log.go new file mode 100644 index 000000000..3a609d76d --- /dev/null +++ b/watchtower/wtdb/migration4/log.go @@ -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 +} diff --git a/watchtower/wtdb/migration4/range_index.go b/watchtower/wtdb/migration4/range_index.go new file mode 100644 index 000000000..79793ce24 --- /dev/null +++ b/watchtower/wtdb/migration4/range_index.go @@ -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 +} diff --git a/watchtower/wtdb/range_index.go b/watchtower/wtdb/range_index.go new file mode 100644 index 000000000..77d09f645 --- /dev/null +++ b/watchtower/wtdb/range_index.go @@ -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 +} diff --git a/watchtower/wtdb/range_index_test.go b/watchtower/wtdb/range_index_test.go new file mode 100644 index 000000000..b03b9a410 --- /dev/null +++ b/watchtower/wtdb/range_index_test.go @@ -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 +} diff --git a/watchtower/wtdb/version.go b/watchtower/wtdb/version.go index 4785b0ae2..ebc09fb8d 100644 --- a/watchtower/wtdb/version.go +++ b/watchtower/wtdb/version.go @@ -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 } diff --git a/watchtower/wtmock/client_db.go b/watchtower/wtmock/client_db.go index b12fe2780..095e8cbac 100644 --- a/watchtower/wtmock/client_db.go +++ b/watchtower/wtmock/client_db.go @@ -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 +}