12250203495

Committed 10 Dec 2024 05:45AM UTC coverage: 49.842% (+0.07%) from 49.773%

Build # 12250203495

Build Type

Pull #9342

github

Committed by

ellemouton

Commit Message

docs: add release notes entry

Pull Request Pull Request #9342: protofsm: update GR Manager usage and start using structured logging

Run Details

100221 of 201077 relevant lines covered (49.84%)

2.07 hits per line

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

58.59

/watchtower/wtdb/client_db.go

package wtdb

import (
        "bytes"
        "errors"
        "fmt"
        "math"
        "net"
        "sync"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/tlv"
        "github.com/lightningnetwork/lnd/watchtower/blob"
)

var (
        // cSessionKeyIndexBkt is a top-level bucket storing:
        //   tower-id -> reserved-session-key-index (uint32).
        cSessionKeyIndexBkt = []byte("client-session-key-index-bucket")

        // cChanDetailsBkt is a top-level bucket storing:
        //   channel-id => cChannelSummary -> encoded ClientChanSummary.
        //                  => cChanDBID -> db-assigned-id
        //                 => cChanSessions => db-session-id -> 1
        //                 => cChanClosedHeight -> block-height
        //                 => cChanMaxCommitmentHeight -> commitment-height
        cChanDetailsBkt = []byte("client-channel-detail-bucket")

        // cChanSessions is a sub-bucket of cChanDetailsBkt which stores:
        //    db-session-id -> 1
        cChanSessions = []byte("client-channel-sessions")

        // cChanDBID is a key used in the cChanDetailsBkt to store the
        // db-assigned-id of a channel.
        cChanDBID = []byte("client-channel-db-id")

        // cChanClosedHeight is a key used in the cChanDetailsBkt to store the
        // block height at which the channel's closing transaction was mined in.
        // If this there is no associated value for this key, then the channel
        // has not yet been marked as closed.
        cChanClosedHeight = []byte("client-channel-closed-height")

        // cChannelSummary is a key used in cChanDetailsBkt to store the encoded
        // body of ClientChanSummary.
        cChannelSummary = []byte("client-channel-summary")

        // cChanMaxCommitmentHeight is a key used in the cChanDetailsBkt used
        // to store the highest commitment height for this channel that the
        // tower has been handed.
        cChanMaxCommitmentHeight = []byte(
                "client-channel-max-commitment-height",
        )

        // cSessionBkt is a top-level bucket storing:
        //   session-id => cSessionBody -> encoded ClientSessionBody
        //                 => cSessionDBID -> db-assigned-id
        //              => cSessionCommits => seqnum -> encoded CommittedUpdate
        //              => cSessionAckRangeIndex => db-chan-id => start -> end
        //                 => cSessionRogueUpdateCount -> count
        cSessionBkt = []byte("client-session-bucket")

        // cSessionDBID is a key used in the cSessionBkt to store the
        // db-assigned-id of a session.
        cSessionDBID = []byte("client-session-db-id")

        // cSessionBody is a sub-bucket of cSessionBkt storing only the body of
        // the ClientSession.
        cSessionBody = []byte("client-session-body")

        // cSessionBody is a sub-bucket of cSessionBkt storing:
        //    seqnum -> encoded CommittedUpdate.
        cSessionCommits = []byte("client-session-commits")

        // cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing
        //    chan-id => start -> end
        cSessionAckRangeIndex = []byte("client-session-ack-range-index")

        // cSessionRogueUpdateCount is a key in the cSessionBkt bucket storing
        // the number of rogue updates that were backed up using the session.
        // Rogue updates are updates for channels that have been closed already
        // at the time of the back-up.
        cSessionRogueUpdateCount = []byte("client-session-rogue-update-count")

        // cChanIDIndexBkt is a top-level bucket storing:
        //    db-assigned-id -> channel-ID
        cChanIDIndexBkt = []byte("client-channel-id-index")

        // cSessionIDIndexBkt is a top-level bucket storing:
        //    db-assigned-id -> session-id
        cSessionIDIndexBkt = []byte("client-session-id-index")

        // cTowerBkt is a top-level bucket storing:
        //    tower-id -> encoded Tower.
        cTowerBkt = []byte("client-tower-bucket")

        // cTowerIndexBkt is a top-level bucket storing:
        //    tower-pubkey -> tower-id.
        cTowerIndexBkt = []byte("client-tower-index-bucket")

        // cTowerToSessionIndexBkt is a top-level bucket storing:
        //         tower-id -> session-id -> 1
        cTowerToSessionIndexBkt = []byte(
                "client-tower-to-session-index-bucket",
        )

        // cClosableSessionsBkt is a top-level bucket storing:
        //         db-session-id -> last-channel-close-height
        cClosableSessionsBkt = []byte("client-closable-sessions-bucket")

        // cTaskQueue is a top-level bucket where the disk queue may store its
        // content.
        cTaskQueue = []byte("client-task-queue")

        // ErrTowerNotFound signals that the target tower was not found in the
        // database.
        ErrTowerNotFound = errors.New("tower not found")

        // ErrTowerUnackedUpdates is an error returned when we attempt to mark a
        // tower's sessions as inactive, but one of its sessions has unacked
        // updates.
        ErrTowerUnackedUpdates = errors.New("tower has unacked updates")

        // ErrCorruptClientSession signals that the client session's on-disk
        // 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(
                "client session already exists",
        )

        // ErrChannelAlreadyRegistered signals a duplicate attempt to register a
        // channel with the client database.
        ErrChannelAlreadyRegistered = errors.New("channel already registered")

        // ErrChannelNotRegistered signals a channel has not yet been registered
        // in the client database.
        ErrChannelNotRegistered = errors.New("channel not registered")

        // ErrClientSessionNotFound signals that the requested client session
        // was not found in the database.
        ErrClientSessionNotFound = errors.New("client session not found")

        // ErrUpdateAlreadyCommitted signals that the chosen sequence number has
        // already been committed to an update with a different breach hint.
        ErrUpdateAlreadyCommitted = errors.New("update already committed")

        // ErrCommitUnorderedUpdate signals the client tried to commit a
        // sequence number other than the next unallocated sequence number.
        ErrCommitUnorderedUpdate = errors.New("update seqnum not monotonic")

        // ErrCommittedUpdateNotFound signals that the tower tried to ACK a
        // sequence number that has not yet been allocated by the client.
        ErrCommittedUpdateNotFound = errors.New("committed update not found")

        // ErrUnallocatedLastApplied signals that the tower tried to provide a
        // LastApplied value greater than any allocated sequence number.
        ErrUnallocatedLastApplied = errors.New("tower echoed last appiled " +
                "greater than allocated seqnum")

        // ErrNoReservedKeyIndex signals that a client session could not be
        // created because no session key index was reserved.
        ErrNoReservedKeyIndex = errors.New("key index not reserved")

        // ErrIncorrectKeyIndex signals that the client session could not be
        // created because session key index differs from the reserved key
        // index.
        ErrIncorrectKeyIndex = errors.New("incorrect key index")

        // 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")

        // ErrSessionFailedFilterFn indicates that a particular session did
        // not pass the filter func provided by the caller.
        ErrSessionFailedFilterFn = errors.New("session failed filter func")

        // ErrSessionNotClosable is returned when a session is not found in the
        // closable list.
        ErrSessionNotClosable = errors.New("session is not closable")

        // errSessionHasOpenChannels is an error used to indicate that a
        // session has updates for channels that are still open.
        errSessionHasOpenChannels = errors.New("session has open channels")

        // ErrSessionHasUnackedUpdates is an error used to indicate that a
        // session has un-acked updates.
        ErrSessionHasUnackedUpdates = errors.New("session has un-acked updates")

        // errChannelHasMoreSessions is an error used to indicate that a channel
        // has updates in other non-closed sessions.
        errChannelHasMoreSessions = errors.New("channel has updates in " +
                "other sessions")
)

// NewBoltBackendCreator returns a function that creates a new bbolt backend for
// the watchtower database.
func NewBoltBackendCreator(active bool, dbPath,
        dbFileName string) func(boltCfg *kvdb.BoltConfig) (kvdb.Backend,
        error) {

        // If the watchtower client isn't active, we return a function that
        // always returns a nil DB to make sure we don't create empty database
        // files.
        if !active {
                return func(_ *kvdb.BoltConfig) (kvdb.Backend, error) {
                        return nil, nil
                }
        }

        return func(boltCfg *kvdb.BoltConfig) (kvdb.Backend, error) {
                cfg := &kvdb.BoltBackendConfig{
                        DBPath:            dbPath,
                        DBFileName:        dbFileName,
                        NoFreelistSync:    boltCfg.NoFreelistSync,
                        AutoCompact:       boltCfg.AutoCompact,
                        AutoCompactMinAge: boltCfg.AutoCompactMinAge,
                        DBTimeout:         boltCfg.DBTimeout,
                }

                db, err := kvdb.GetBoltBackend(cfg)
                if err != nil {
                        return nil, fmt.Errorf("could not open boltdb: %w", err)
                }

                return db, nil
        }
}

// ClientDB is single database providing a persistent storage engine for the
// 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
// directory. If no such database exists, this method will initialize a fresh
// one using the latest version number and bucket structure. If a database
// exists but has a lower version number than the current version, any necessary
// migrations will be applied before returning. Any attempt to open a database
// with a version number higher that the latest version will fail to prevent
// accidental reversion.
func OpenClientDB(db kvdb.Backend) (*ClientDB, error) {
        firstInit, err := isFirstInit(db)
        if err != nil {
                return nil, err
        }

        clientDB := &ClientDB{
                db: db,
                ackedRangeIndex: make(
                        map[SessionID]map[lnwire.ChannelID]*RangeIndex,
                ),
        }

        err = initOrSyncVersions(clientDB, firstInit, clientDBVersions)
        if err != nil {
                db.Close()
                return nil, err
        }

        // Now that the database version fully consistent with our latest known
        // version, ensure that all top-level buckets known to this version are
        // initialized. This allows us to assume their presence throughout all
        // operations. If an known top-level bucket is expected to exist but is
        // missing, this will trigger a ErrUninitializedDB error.
        err = kvdb.Update(clientDB.db, initClientDBBuckets, func() {})
        if err != nil {
                db.Close()
                return nil, err
        }

        return clientDB, nil
}

// initClientDBBuckets creates all top-level buckets required to handle database
// operations required by the latest version.
func initClientDBBuckets(tx kvdb.RwTx) error {
        buckets := [][]byte{
                cSessionKeyIndexBkt,
                cChanDetailsBkt,
                cSessionBkt,
                cTowerBkt,
                cTowerIndexBkt,
                cTowerToSessionIndexBkt,
                cChanIDIndexBkt,
                cSessionIDIndexBkt,
                cClosableSessionsBkt,
        }

        for _, bucket := range buckets {
                _, err := tx.CreateTopLevelBucket(bucket)
                if err != nil {
                        return err
                }
        }

        return nil
}

// bdb returns the backing bbolt.DB instance.
//
// NOTE: Part of the versionedDB interface.
func (c *ClientDB) bdb() kvdb.Backend {
        return c.db
}

// Version returns the database's current version number.
//
// NOTE: Part of the versionedDB interface.
func (c *ClientDB) Version() (uint32, error) {
        var version uint32
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                var err error
                version, err = getDBVersion(tx)
                return err
        }, func() {
                version = 0
        })
        if err != nil {
                return 0, err
        }

        return version, nil
}

// Close closes the underlying database.
func (c *ClientDB) Close() error {
        return c.db.Close()
}

// CreateTower initialize an address record used to communicate with a
// watchtower. Each Tower is assigned a unique ID, that is used to amortize
// storage costs of the public key when used by multiple sessions. If the tower
// already exists, the address is appended to the list of all addresses used to
// that tower previously and its corresponding sessions are marked as active.
func (c *ClientDB) CreateTower(lnAddr *lnwire.NetAddress) (*Tower, error) {
        var towerPubKey [33]byte
        copy(towerPubKey[:], lnAddr.IdentityKey.SerializeCompressed())

        var tower *Tower
        err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                towerIndex := tx.ReadWriteBucket(cTowerIndexBkt)
                if towerIndex == nil {
                        return ErrUninitializedDB
                }

                towers := tx.ReadWriteBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }

                towerToSessionIndex := tx.ReadWriteBucket(
                        cTowerToSessionIndexBkt,
                )
                if towerToSessionIndex == nil {
                        return ErrUninitializedDB
                }

                // Check if the tower index already knows of this pubkey.
                towerIDBytes := towerIndex.Get(towerPubKey[:])
                if len(towerIDBytes) == 8 {
                        // The tower already exists, deserialize the existing
                        // record.
                        var err error
                        tower, err = getTower(towers, towerIDBytes)
                        if err != nil {
                                return err
                        }

                        // Set its status to active.
                        tower.Status = TowerStatusActive

                        // Add the new address to the existing tower. If the
                        // address is a duplicate, this will result in no
                        // change.
                        tower.AddAddress(lnAddr.Address)
                } else {
                        // No such tower exists, create a new tower id for our
                        // new tower. The error is unhandled since NextSequence
                        // never fails in an Update.
                        towerID, _ := towerIndex.NextSequence()

                        tower = &Tower{
                                ID:          TowerID(towerID),
                                IdentityKey: lnAddr.IdentityKey,
                                Addresses:   []net.Addr{lnAddr.Address},
                                Status:      TowerStatusActive,
                        }

                        towerIDBytes = tower.ID.Bytes()

                        // Since this tower is new, record the mapping from
                        // tower pubkey to tower id in the tower index.
                        err := towerIndex.Put(towerPubKey[:], towerIDBytes)
                        if err != nil {
                                return err
                        }

                        // Create a new bucket for this tower in the
                        // tower-to-sessions index.
                        _, err = towerToSessionIndex.CreateBucket(towerIDBytes)
                        if err != nil {
                                return err
                        }
                }

                // Store the new or updated tower under its tower id.
                return putTower(towers, tower)
        }, func() {
                tower = nil
        })
        if err != nil {
                return nil, err
        }

        return tower, nil
}

// RemoveTower modifies a tower's record within the database. If an address is
// provided, then _only_ the address record should be removed from the tower's
// persisted state. Otherwise, we'll attempt to mark the tower as inactive. If
// any of its sessions has unacked updates, then ErrTowerUnackedUpdates is
// returned. If the tower doesn't have any sessions at all, it'll be completely
// removed from the database.
//
// NOTE: An error is not returned if the tower doesn't exist.
func (c *ClientDB) RemoveTower(pubKey *btcec.PublicKey, addr net.Addr) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                towers := tx.ReadWriteBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }

                towerIndex := tx.ReadWriteBucket(cTowerIndexBkt)
                if towerIndex == nil {
                        return ErrUninitializedDB
                }

                towersToSessionsIndex := tx.ReadWriteBucket(
                        cTowerToSessionIndexBkt,
                )
                if towersToSessionsIndex == nil {
                        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()
                towerIDBytes := towerIndex.Get(pubKeyBytes)
                if towerIDBytes == nil {
                        return nil
                }

                tower, err := getTower(towers, towerIDBytes)
                if err != nil {
                        return err
                }

                // If an address is provided, then we should _only_ remove the
                // address record from the database.
                if addr != nil {
                        // Towers should always have at least one address saved.
                        tower.RemoveAddress(addr)
                        if len(tower.Addresses) == 0 {
                                return ErrLastTowerAddr
                        }

                        return putTower(towers, tower)
                }

                // Otherwise, we should attempt to mark the tower's sessions as
                // inactive.
                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }
                towerID := TowerIDFromBytes(towerIDBytes)

                committedUpdateCount := make(map[SessionID]uint16)
                perCommittedUpdate := func(s *ClientSession,
                        _ *CommittedUpdate) {

                        committedUpdateCount[s.ID]++
                }

                towerSessions, err := c.listTowerSessions(
                        towerID, sessions, chanIDIndexBkt,
                        towersToSessionsIndex,
                        WithPerCommittedUpdate(perCommittedUpdate),
                )
                if err != nil {
                        return err
                }

                // If it doesn't have any, we can completely remove it from the
                // database.
                if len(towerSessions) == 0 {
                        if err := towerIndex.Delete(pubKeyBytes); err != nil {
                                return err
                        }

                        if err := towers.Delete(towerIDBytes); err != nil {
                                return err
                        }

                        return towersToSessionsIndex.DeleteNestedBucket(
                                towerIDBytes,
                        )
                }

                // Otherwise, we mark the tower as inactive.
                tower.Status = TowerStatusInactive
                err = putTower(towers, tower)
                if err != nil {
                        return err
                }

                // We'll do a check to ensure that the tower's sessions don't
                // have any pending back-ups.
                for _, session := range towerSessions {
                        if committedUpdateCount[session.ID] > 0 {
                                return ErrTowerUnackedUpdates
                        }
                }

                return nil
        }, func() {})
}

// DeactivateTower sets the given tower's status to inactive. This means that
// this tower's sessions won't be loaded and used for backups. CreateTower can
// be used to reactivate the tower again.
func (c *ClientDB) DeactivateTower(pubKey *btcec.PublicKey) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                towers := tx.ReadWriteBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }

                towerIndex := tx.ReadWriteBucket(cTowerIndexBkt)
                if towerIndex == nil {
                        return ErrUninitializedDB
                }

                towersToSessionsIndex := tx.ReadWriteBucket(
                        cTowerToSessionIndexBkt,
                )
                if towersToSessionsIndex == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                pubKeyBytes := pubKey.SerializeCompressed()
                towerIDBytes := towerIndex.Get(pubKeyBytes)
                if towerIDBytes == nil {
                        return ErrTowerNotFound
                }

                tower, err := getTower(towers, towerIDBytes)
                if err != nil {
                        return err
                }

                // If the tower already has the desired status, then we can exit
                // here.
                if tower.Status == TowerStatusInactive {
                        return nil
                }

                // Otherwise, we update the status and re-store the tower.
                tower.Status = TowerStatusInactive

                return putTower(towers, tower)
        }, func() {})
}

// LoadTowerByID retrieves a tower by its tower ID.
func (c *ClientDB) LoadTowerByID(towerID TowerID) (*Tower, error) {
        var tower *Tower
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                towers := tx.ReadBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }

                var err error
                tower, err = getTower(towers, towerID.Bytes())
                return err
        }, func() {
                tower = nil
        })
        if err != nil {
                return nil, err
        }

        return tower, nil
}

// LoadTower retrieves a tower by its public key.
func (c *ClientDB) LoadTower(pubKey *btcec.PublicKey) (*Tower, error) {
        var tower *Tower
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                towers := tx.ReadBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }
                towerIndex := tx.ReadBucket(cTowerIndexBkt)
                if towerIndex == nil {
                        return ErrUninitializedDB
                }

                towerIDBytes := towerIndex.Get(pubKey.SerializeCompressed())
                if towerIDBytes == nil {
                        return ErrTowerNotFound
                }

                var err error
                tower, err = getTower(towers, towerIDBytes)
                return err
        }, func() {
                tower = nil
        })
        if err != nil {
                return nil, err
        }

        return tower, nil
}

// TowerFilterFn is the signature of a call-back function that can be used to
// skip certain towers in the ListTowers method.
type TowerFilterFn func(*Tower) bool

// ListTowers retrieves the list of towers available within the database that
// have a status matching the given status. The filter function may be set in
// order to filter out the towers to be returned.
func (c *ClientDB) ListTowers(filter TowerFilterFn) ([]*Tower, error) {
        var towers []*Tower
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                towerBucket := tx.ReadBucket(cTowerBkt)
                if towerBucket == nil {
                        return ErrUninitializedDB
                }

                return towerBucket.ForEach(func(towerIDBytes, _ []byte) error {
                        tower, err := getTower(towerBucket, towerIDBytes)
                        if err != nil {
                                return err
                        }

                        if filter != nil && !filter(tower) {
                                return nil
                        }

                        towers = append(towers, tower)

                        return nil
                })
        }, func() {
                towers = nil
        })
        if err != nil {
                return nil, err
        }

        return towers, nil
}

// NextSessionKeyIndex reserves a new session key derivation index for a
// particular tower id. The index is reserved for that tower until
// CreateClientSession is invoked for that tower and index, at which point a new
// index for that tower can be reserved. Multiple calls to this method before
// CreateClientSession is invoked should return the same index unless forceNext
// is true.
func (c *ClientDB) NextSessionKeyIndex(towerID TowerID,
        blobType blob.Type, forceNext bool) (uint32, error) {

        var index uint32
        err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                keyIndex := tx.ReadWriteBucket(cSessionKeyIndexBkt)
                if keyIndex == nil {
                        return ErrUninitializedDB
                }

                var err error
                if !forceNext {
                        // Check the session key index to see if a key has
                        // already been reserved for this tower. If so, we'll
                        // deserialize and return the index directly.
                        index, err = getSessionKeyIndex(
                                keyIndex, towerID, blobType,
                        )
                        if err == nil {
                                return nil
                        }
                }

                // By default, we use the next available bucket sequence as the
                // key index. But if forceNext is true, then it is assumed that
                // some data loss occurred and so the sequence is incremented a
                // by a jump of 1000 so that we can arrive at a brand new key
                // index quicker.
                currentSequence := keyIndex.Sequence()
                nextIndex := currentSequence + 1
                if forceNext {
                        nextIndex = currentSequence + 1000
                }

                if err = keyIndex.SetSequence(nextIndex); err != nil {
                        return fmt.Errorf("could not set next bucket "+
                                "sequence: %w", err)
                }

                // As a sanity check, assert that the index is still in the
                // valid range of unhardened pubkeys. In the future, we should
                // move to only using hardened keys, and this will prevent any
                // overlap from occurring until then. This also prevents us from
                // overflowing uint32s.
                if nextIndex > math.MaxInt32 {
                        return fmt.Errorf("exhausted session key indexes")
                }

                // Create the key that will used to be store the reserved index.
                keyBytes := createSessionKeyIndexKey(towerID, blobType)

                index = uint32(nextIndex)

                var indexBuf [4]byte
                byteOrder.PutUint32(indexBuf[:], index)

                // Record the reserved session key index under this tower's id.
                return keyIndex.Put(keyBytes, indexBuf[:])
        }, func() {
                index = 0
        })
        if err != nil {
                return 0, err
        }

        return index, nil
}

// CreateClientSession records a newly negotiated client session in the set of
// active sessions. The session can be identified by its SessionID.
func (c *ClientDB) CreateClientSession(session *ClientSession) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                keyIndexes := tx.ReadWriteBucket(cSessionKeyIndexBkt)
                if keyIndexes == nil {
                        return ErrUninitializedDB
                }

                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                towers := tx.ReadBucket(cTowerBkt)
                if towers == nil {
                        return ErrUninitializedDB
                }

                towerToSessionIndex := tx.ReadWriteBucket(
                        cTowerToSessionIndexBkt,
                )
                if towerToSessionIndex == nil {
                        return ErrUninitializedDB
                }

                // Check that  client session with this session id doesn't
                // already exist.
                existingSessionBytes := sessions.NestedReadWriteBucket(
                        session.ID[:],
                )
                if existingSessionBytes != nil {
                        return ErrClientSessionAlreadyExists
                }

                // Ensure that a tower with the given ID actually exists in the
                // DB.
                towerID := session.TowerID
                if _, err := getTower(towers, towerID.Bytes()); err != nil {
                        return err
                }

                blobType := session.Policy.BlobType

                // Check that this tower has a reserved key index.
                index, err := getSessionKeyIndex(keyIndexes, towerID, blobType)
                if err != nil {
                        return err
                }

                // Assert that the key index of the inserted session matches the
                // reserved session key index.
                if index != session.KeyIndex {
                        return ErrIncorrectKeyIndex
                }

                // Remove the key index reservation. For altruist commit
                // sessions, we'll also purge under the old legacy key format.
                key := createSessionKeyIndexKey(towerID, blobType)
                err = keyIndexes.Delete(key)
                if err != nil {
                        return err
                }
                if blobType == blob.TypeAltruistCommit {
                        err = keyIndexes.Delete(towerID.Bytes())
                        if err != nil {
                                return err
                        }
                }

                // Get the session-ID index bucket.
                dbIDIndex := tx.ReadWriteBucket(cSessionIDIndexBkt)
                if dbIDIndex == nil {
                        return ErrUninitializedDB
                }

                // Get a new, unique, ID for this session from the session-ID
                // index bucket.
                nextSeq, err := dbIDIndex.NextSequence()
                if err != nil {
                        return err
                }

                // Add the new entry to the dbID-to-SessionID index.
                newIndex, err := writeBigSize(nextSeq)
                if err != nil {
                        return err
                }

                err = dbIDIndex.Put(newIndex, session.ID[:])
                if err != nil {
                        return err
                }

                // Also add the db-assigned-id to the session bucket under the
                // cSessionDBID key.
                sessionBkt, err := sessions.CreateBucket(session.ID[:])
                if err != nil {
                        return err
                }

                err = sessionBkt.Put(cSessionDBID, newIndex)
                if err != nil {
                        return err
                }

                // TODO(elle): migrate the towerID-to-SessionID to use the
                // new db-assigned sessionID's rather.

                // Add the new entry to the towerID-to-SessionID index.
                towerSessions := towerToSessionIndex.NestedReadWriteBucket(
                        towerID.Bytes(),
                )
                if towerSessions == nil {
                        return ErrTowerNotFound
                }

                err = towerSessions.Put(session.ID[:], []byte{1})
                if err != nil {
                        return err
                }

                // Finally, write the client session's body in the sessions
                // bucket.
                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(sessionBkt kvdb.RwBucket, dbChanIDBytes []byte) (
        kvdb.RwBucket, error) {

        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.
//
// NOTE: The original serialization only used tower-id, which prevents
// concurrent client types from reserving sessions with the same tower.
func createSessionKeyIndexKey(towerID TowerID, blobType blob.Type) []byte {
        towerIDBytes := towerID.Bytes()

        // Session key indexes are stored under as tower-id||blob-type.
        var keyBytes [6]byte
        copy(keyBytes[:4], towerIDBytes)
        byteOrder.PutUint16(keyBytes[4:], uint16(blobType))

        return keyBytes[:]
}

// getSessionKeyIndex is a helper method.
func getSessionKeyIndex(keyIndexes kvdb.RwBucket, towerID TowerID,
        blobType blob.Type) (uint32, error) {

        // Session key indexes are store under as tower-id||blob-type. The
        // original serialization only used tower-id, which prevents concurrent
        // client types from reserving sessions with the same tower.
        keyBytes := createSessionKeyIndexKey(towerID, blobType)

        // Retrieve the index using the key bytes. If the key wasn't found, we
        // will fall back to the legacy format that only uses the tower id, but
        // _only_ if the blob type is for altruist commit sessions since that
        // was the only operational session type prior to changing the key
        // format.
        keyIndexBytes := keyIndexes.Get(keyBytes)
        if keyIndexBytes == nil && blobType == blob.TypeAltruistCommit {
                keyIndexBytes = keyIndexes.Get(towerID.Bytes())
        }

        // All session key indexes should be serialized uint32's. If no key
        // index was found, the length of keyIndexBytes will be 0.
        if len(keyIndexBytes) != 4 {
                return 0, ErrNoReservedKeyIndex
        }

        return byteOrder.Uint32(keyIndexBytes), nil
}

// GetClientSession loads the ClientSession with the given ID from the DB.
func (c *ClientDB) GetClientSession(id SessionID,
        opts ...ClientSessionListOption) (*ClientSession, error) {

        var sess *ClientSession
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                sessionsBkt := tx.ReadBucket(cSessionBkt)
                if sessionsBkt == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                session, err := c.getClientSession(
                        sessionsBkt, chanIDIndexBkt, id[:], opts...,
                )
                if err != nil {
                        return err
                }

                sess = session

                return nil
        }, func() {})

        return sess, err
}

// ListClientSessions returns the set of all client sessions known to the db. An
// optional tower ID can be used to filter out any client sessions in the
// response that do not correspond to this tower.
func (c *ClientDB) ListClientSessions(id *TowerID,
        opts ...ClientSessionListOption) (map[SessionID]*ClientSession, error) {

        var clientSessions map[SessionID]*ClientSession
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                // If no tower ID is specified, then fetch all the sessions
                // known to the db.
                var err error
                if id == nil {
                        clientSessions, err = c.listClientAllSessions(
                                sessions, chanIDIndexBkt, opts...,
                        )
                        return err
                }

                // Otherwise, fetch the sessions for the given tower.
                towerToSessionIndex := tx.ReadBucket(cTowerToSessionIndexBkt)
                if towerToSessionIndex == nil {
                        return ErrUninitializedDB
                }

                clientSessions, err = c.listTowerSessions(
                        *id, sessions, chanIDIndexBkt, towerToSessionIndex,
                        opts...,
                )
                return err
        }, func() {
                clientSessions = nil
        })
        if err != nil {
                return nil, err
        }

        return clientSessions, nil
}

// listClientAllSessions returns the set of all client sessions known to the db.
func (c *ClientDB) listClientAllSessions(sessions, chanIDIndexBkt kvdb.RBucket,
        opts ...ClientSessionListOption) (map[SessionID]*ClientSession, error) {

        clientSessions := make(map[SessionID]*ClientSession)
        err := sessions.ForEach(func(k, _ []byte) error {
                // We'll load the full client session since the client will need
                // the CommittedUpdates and AckedUpdates on startup to resume
                // committed updates and compute the highest known commit height
                // for each channel.
                session, err := c.getClientSession(
                        sessions, chanIDIndexBkt, k, opts...,
                )
                if errors.Is(err, ErrSessionFailedFilterFn) {
                        return nil
                } else if err != nil {
                        return err
                }

                clientSessions[session.ID] = session

                return nil
        })
        if err != nil {
                return nil, err
        }

        return clientSessions, nil
}

// listTowerSessions returns the set of all client sessions known to the db
// that are associated with the given tower id.
func (c *ClientDB) listTowerSessions(id TowerID, sessionsBkt, chanIDIndexBkt,
        towerToSessionIndex kvdb.RBucket, opts ...ClientSessionListOption) (
        map[SessionID]*ClientSession, error) {

        towerIndexBkt := towerToSessionIndex.NestedReadBucket(id.Bytes())
        if towerIndexBkt == nil {
                return nil, ErrTowerNotFound
        }

        clientSessions := make(map[SessionID]*ClientSession)
        err := towerIndexBkt.ForEach(func(k, _ []byte) error {
                // We'll load the full client session since the client will need
                // the CommittedUpdates and AckedUpdates on startup to resume
                // committed updates and compute the highest known commit height
                // for each channel.
                session, err := c.getClientSession(
                        sessionsBkt, chanIDIndexBkt, k, opts...,
                )
                if errors.Is(err, ErrSessionFailedFilterFn) {
                        return nil
                } else if err != nil {
                        return err
                }

                clientSessions[session.ID] = session
                return nil
        })
        if err != nil {
                return nil, err
        }

        return clientSessions, nil
}

// FetchSessionCommittedUpdates retrieves the current set of un-acked updates
// of the given session.
func (c *ClientDB) FetchSessionCommittedUpdates(id *SessionID) (
        []CommittedUpdate, error) {

        var committedUpdates []CommittedUpdate
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                sessionBkt := sessions.NestedReadBucket(id[:])
                if sessionBkt == nil {
                        return ErrClientSessionNotFound
                }

                var err error
                committedUpdates, err = getClientSessionCommits(
                        sessionBkt, nil, nil,
                )
                return err
        }, func() {})
        if err != nil {
                return nil, err
        }

        return committedUpdates, nil
}

// 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 {
                sessions := tx.ReadBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                sessionBkt := sessions.NestedReadBucket(id[:])
                if sessionBkt == nil {
                        return nil
                }

                // First, account for any rogue updates.
                rogueCountBytes := sessionBkt.Get(cSessionRogueUpdateCount)
                if len(rogueCountBytes) != 0 {
                        rogueCount, err := readBigSize(rogueCountBytes)
                        if err != nil {
                                return err
                        }

                        numAcked += rogueCount
                }

                // Then, check if the session-ack-ranges contains any entries
                // to account for.
                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
                        }

                        numAcked += index.NumInSet()

                        return nil
                })
        }, func() {
                numAcked = 0
        })
        if err != nil {
                return 0, err
        }

        return numAcked, nil
}

// FetchChanInfos loads a mapping from all registered channels to their
// ChannelInfo. Only the channels that have not yet been marked as closed will
// be loaded.
func (c *ClientDB) FetchChanInfos() (ChannelInfos, error) {
        var infos ChannelInfos

        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
                        }
                        // If this channel has already been marked as closed,
                        // then its summary does not need to be loaded.
                        closedHeight := chanDetails.Get(cChanClosedHeight)
                        if len(closedHeight) > 0 {
                                return nil
                        }
                        var chanID lnwire.ChannelID
                        copy(chanID[:], k)
                        summary, err := getChanSummary(chanDetails)
                        if err != nil {
                                return err
                        }

                        info := &ChannelInfo{
                                ClientChanSummary: *summary,
                        }

                        maxHeightBytes := chanDetails.Get(
                                cChanMaxCommitmentHeight,
                        )
                        if len(maxHeightBytes) != 0 {
                                height, err := readBigSize(maxHeightBytes)
                                if err != nil {
                                        return err
                                }

                                info.MaxHeight = fn.Some(height)
                        }

                        infos[chanID] = info

                        return nil
                })
        }, func() {
                infos = make(ChannelInfos)
        })
        if err != nil {
                return nil, err
        }

        return infos, nil
}

// RegisterChannel registers a channel for use within the client database. For
// now, all that is stored in the channel summary is the sweep pkscript that
// we'd like any tower sweeps to pay into. In the future, this will be extended
// to contain more info to allow the client efficiently request historical
// states to be backed up under the client's active policy.
func (c *ClientDB) RegisterChannel(chanID lnwire.ChannelID,
        sweepPkScript []byte) error {

        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
                if chanDetailsBkt == nil {
                        return ErrUninitializedDB
                }

                chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID[:])
                if chanDetails != nil {
                        // Channel is already registered.
                        return ErrChannelAlreadyRegistered
                }

                chanDetails, err := chanDetailsBkt.CreateBucket(chanID[:])
                if err != nil {
                        return err
                }

                // 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
                }

                summary := ClientChanSummary{
                        SweepPkScript: sweepPkScript,
                }

                return putChanSummary(chanDetails, &summary)
        }, func() {})
}

// MarkBackupIneligible records that the state identified by the (channel id,
// commit height) tuple was ineligible for being backed up under the current
// policy. This state can be retried later under a different policy.
func (c *ClientDB) MarkBackupIneligible(chanID lnwire.ChannelID,
        commitHeight uint64) error {

        return nil
}

// ListClosableSessions fetches and returns the IDs for all sessions marked as
// closable.
func (c *ClientDB) ListClosableSessions() (map[SessionID]uint32, error) {
        sessions := make(map[SessionID]uint32)
        err := kvdb.View(c.db, func(tx kvdb.RTx) error {
                csBkt := tx.ReadBucket(cClosableSessionsBkt)
                if csBkt == nil {
                        return ErrUninitializedDB
                }

                sessIDIndexBkt := tx.ReadBucket(cSessionIDIndexBkt)
                if sessIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                return csBkt.ForEach(func(dbIDBytes, heightBytes []byte) error {
                        dbID, err := readBigSize(dbIDBytes)
                        if err != nil {
                                return err
                        }

                        sessID, err := getRealSessionID(sessIDIndexBkt, dbID)
                        if err != nil {
                                return err
                        }

                        sessions[*sessID] = byteOrder.Uint32(heightBytes)

                        return nil
                })
        }, func() {
                sessions = make(map[SessionID]uint32)
        })
        if err != nil {
                return nil, err
        }

        return sessions, nil
}

// DeleteSession can be called when a session should be deleted from the DB.
// All references to the session will also be deleted from the DB. Note that a
// session will only be deleted if was previously marked as closable.
func (c *ClientDB) DeleteSession(id SessionID) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessionsBkt := tx.ReadWriteBucket(cSessionBkt)
                if sessionsBkt == nil {
                        return ErrUninitializedDB
                }

                closableBkt := tx.ReadWriteBucket(cClosableSessionsBkt)
                if closableBkt == nil {
                        return ErrUninitializedDB
                }

                chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
                if chanDetailsBkt == nil {
                        return ErrUninitializedDB
                }

                sessIDIndexBkt := tx.ReadWriteBucket(cSessionIDIndexBkt)
                if sessIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadWriteBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                towerToSessBkt := tx.ReadWriteBucket(cTowerToSessionIndexBkt)
                if towerToSessBkt == nil {
                        return ErrUninitializedDB
                }

                // Get the sub-bucket for this session ID. If it does not exist
                // then the session has already been deleted and so our work is
                // done.
                sessionBkt := sessionsBkt.NestedReadBucket(id[:])
                if sessionBkt == nil {
                        return nil
                }

                _, dbIDBytes, err := getDBSessionID(sessionsBkt, id)
                if err != nil {
                        return err
                }

                // First we check if the session has actually been marked as
                // closable.
                if closableBkt.Get(dbIDBytes) == nil {
                        return ErrSessionNotClosable
                }

                sess, err := getClientSessionBody(sessionsBkt, id[:])
                if err != nil {
                        return err
                }

                // Delete from the tower-to-sessionID index.
                towerIndexBkt := towerToSessBkt.NestedReadWriteBucket(
                        sess.TowerID.Bytes(),
                )
                if towerIndexBkt == nil {
                        return fmt.Errorf("no entry in the tower-to-session "+
                                "index found for tower ID %v", sess.TowerID)
                }

                err = towerIndexBkt.Delete(id[:])
                if err != nil {
                        return err
                }

                // Delete entry from session ID index.
                err = sessIDIndexBkt.Delete(dbIDBytes)
                if err != nil {
                        return err
                }

                // Delete the entry from the closable sessions index.
                err = closableBkt.Delete(dbIDBytes)
                if err != nil {
                        return err
                }

                ackRanges := sessionBkt.NestedReadBucket(cSessionAckRangeIndex)

                // There is a small chance that the session only contains rogue
                // updates. In that case, there will be no ack-ranges index but
                // the rogue update count will be equal the MaxUpdates.
                rogueCountBytes := sessionBkt.Get(cSessionRogueUpdateCount)
                if len(rogueCountBytes) != 0 {
                        rogueCount, err := readBigSize(rogueCountBytes)
                        if err != nil {
                                return err
                        }

                        maxUpdates := sess.ClientSessionBody.Policy.MaxUpdates
                        if rogueCount == uint64(maxUpdates) {
                                // Do a sanity check to ensure that the acked
                                // ranges bucket does not exist in this case.
                                if ackRanges != nil {
                                        return fmt.Errorf("acked updates "+
                                                "exist for session with a "+
                                                "max-updates(%d) rogue count",
                                                rogueCount)
                                }

                                return sessionsBkt.DeleteNestedBucket(id[:])
                        }
                }

                // A session would only be considered closable if it was
                // exhausted. Meaning that it should not be the case that it has
                // no acked-updates.
                if ackRanges == nil {
                        return fmt.Errorf("cannot delete session %s since it "+
                                "is not yet exhausted", id)
                }

                // For each of the channels, delete the session ID entry.
                err = ackRanges.ForEach(func(chanDBID, _ []byte) error {
                        chanDBIDInt, err := readBigSize(chanDBID)
                        if err != nil {
                                return err
                        }

                        chanID, err := getRealChannelID(
                                chanIDIndexBkt, chanDBIDInt,
                        )
                        if err != nil {
                                return err
                        }

                        chanDetails := chanDetailsBkt.NestedReadWriteBucket(
                                chanID[:],
                        )
                        if chanDetails == nil {
                                return ErrChannelNotRegistered
                        }

                        chanSessions := chanDetails.NestedReadWriteBucket(
                                cChanSessions,
                        )
                        if chanSessions == nil {
                                return fmt.Errorf("no session list found for "+
                                        "channel %s", chanID)
                        }

                        // Check that this session was actually listed in the
                        // session list for this channel.
                        if len(chanSessions.Get(dbIDBytes)) == 0 {
                                return fmt.Errorf("session %s not found in "+
                                        "the session list for channel %s", id,
                                        chanID)
                        }

                        // If it was, then delete it.
                        err = chanSessions.Delete(dbIDBytes)
                        if err != nil {
                                return err
                        }

                        // If this was the last session for this channel, we can
                        // now delete the channel details for this channel
                        // completely.
                        err = chanSessions.ForEach(func(_, _ []byte) error {
                                return errChannelHasMoreSessions
                        })
                        if errors.Is(err, errChannelHasMoreSessions) {
                                return nil
                        } else if err != nil {
                                return err
                        }

                        // Delete the channel's entry from the channel-id-index.
                        dbID := chanDetails.Get(cChanDBID)
                        err = chanIDIndexBkt.Delete(dbID)
                        if err != nil {
                                return err
                        }

                        // Delete the channel details.
                        return chanDetailsBkt.DeleteNestedBucket(chanID[:])
                })
                if err != nil {
                        return err
                }

                // Delete the actual session.
                return sessionsBkt.DeleteNestedBucket(id[:])
        }, func() {})
}

// MarkChannelClosed will mark a registered channel as closed by setting its
// closed-height as the given block height. It returns a list of session IDs for
// sessions that are now considered closable due to the close of this channel.
// The details for this channel will be deleted from the DB if there are no more
// sessions in the DB that contain updates for this channel.
func (c *ClientDB) MarkChannelClosed(chanID lnwire.ChannelID,
        blockHeight uint32) ([]SessionID, error) {

        var closableSessions []SessionID
        err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessionsBkt := tx.ReadBucket(cSessionBkt)
                if sessionsBkt == nil {
                        return ErrUninitializedDB
                }

                chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
                if chanDetailsBkt == nil {
                        return ErrUninitializedDB
                }

                closableSessBkt := tx.ReadWriteBucket(cClosableSessionsBkt)
                if closableSessBkt == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                sessIDIndexBkt := tx.ReadBucket(cSessionIDIndexBkt)
                if sessIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID[:])
                if chanDetails == nil {
                        return ErrChannelNotRegistered
                }

                // If there are no sessions for this channel, the channel
                // details can be deleted.
                chanSessIDsBkt := chanDetails.NestedReadBucket(cChanSessions)
                if chanSessIDsBkt == nil {
                        return chanDetailsBkt.DeleteNestedBucket(chanID[:])
                }

                // Otherwise, mark the channel as closed.
                var height [4]byte
                byteOrder.PutUint32(height[:], blockHeight)

                err := chanDetails.Put(cChanClosedHeight, height[:])
                if err != nil {
                        return err
                }

                // Now iterate through all the sessions of the channel to check
                // if any of them are closeable.
                return chanSessIDsBkt.ForEach(func(sessDBID, _ []byte) error {
                        sessDBIDInt, err := readBigSize(sessDBID)
                        if err != nil {
                                return err
                        }

                        // Use the session-ID index to get the real session ID.
                        sID, err := getRealSessionID(
                                sessIDIndexBkt, sessDBIDInt,
                        )
                        if err != nil {
                                return err
                        }

                        isClosable, err := isSessionClosable(
                                sessionsBkt, chanDetailsBkt, chanIDIndexBkt,
                                sID,
                        )
                        if err != nil {
                                return err
                        }

                        if !isClosable {
                                return nil
                        }

                        // Add session to "closableSessions" list and add the
                        // block height that this last channel was closed in.
                        // This will be used in future to determine when we
                        // should delete the session.
                        var height [4]byte
                        byteOrder.PutUint32(height[:], blockHeight)
                        err = closableSessBkt.Put(sessDBID, height[:])
                        if err != nil {
                                return err
                        }

                        closableSessions = append(closableSessions, *sID)

                        return nil
                })
        }, func() {
                closableSessions = nil
        })
        if err != nil {
                return nil, err
        }

        return closableSessions, nil
}

// isSessionClosable returns true if a session is considered closable. A session
// is considered closable only if all the following points are true:
//  1. It has no un-acked updates.
//  2. It is exhausted (ie it can't accept any more updates) OR it has been
//     marked as terminal.
//  3. All the channels that it has acked updates for are closed.
func isSessionClosable(sessionsBkt, chanDetailsBkt, chanIDIndexBkt kvdb.RBucket,
        id *SessionID) (bool, error) {

        sessBkt := sessionsBkt.NestedReadBucket(id[:])
        if sessBkt == nil {
                return false, ErrSessionNotFound
        }

        // Since the DeleteCommittedUpdates method deletes the cSessionCommits
        // bucket in one go, it is possible for the session to be closable even
        // if this bucket no longer exists.
        commitsBkt := sessBkt.NestedReadBucket(cSessionCommits)
        if commitsBkt != nil {
                // If the session has any un-acked updates, then it is not yet
                // closable.
                err := commitsBkt.ForEach(func(_, _ []byte) error {
                        return ErrSessionHasUnackedUpdates
                })
                if errors.Is(err, ErrSessionHasUnackedUpdates) {
                        return false, nil
                } else if err != nil {
                        return false, err
                }
        }

        session, err := getClientSessionBody(sessionsBkt, id[:])
        if err != nil {
                return false, err
        }

        isTerminal := session.Status == CSessionTerminal

        // We have already checked that the session has no more committed
        // updates. So now we can check if the session is exhausted or has a
        // terminal state.
        if !isTerminal && session.SeqNum < session.Policy.MaxUpdates {
                // If the session is not yet exhausted, and it is not yet in a
                // terminal state then it is not yet closable.
                return false, nil
        }

        // Either the acked-update bucket should exist _or_ the rogue update
        // count must be equal to the session's MaxUpdates value, otherwise
        // something is wrong because the above check ensures that the session
        // has been exhausted.
        rogueCountBytes := sessBkt.Get(cSessionRogueUpdateCount)
        if len(rogueCountBytes) != 0 {
                rogueCount, err := readBigSize(rogueCountBytes)
                if err != nil {
                        return false, err
                }

                if rogueCount == uint64(session.Policy.MaxUpdates) {
                        return true, nil
                }
        }

        ackedRangeBkt := sessBkt.NestedReadBucket(cSessionAckRangeIndex)
        if ackedRangeBkt == nil {
                if isTerminal {
                        return true, nil
                }

                // If the session has no acked-updates, and it is not in a
                // terminal state then something is wrong since the above check
                // ensures that this session has been exhausted meaning that it
                // should have MaxUpdates acked updates.
                return false, fmt.Errorf("no acked-updates found for "+
                        "exhausted session %s", id)
        }

        // Iterate over each of the channels that the session has acked-updates
        // for. If any of those channels are not closed, then the session is
        // not yet closable.
        err = ackedRangeBkt.ForEach(func(dbChanID, _ []byte) error {
                dbChanIDInt, err := readBigSize(dbChanID)
                if err != nil {
                        return err
                }

                chanID, err := getRealChannelID(chanIDIndexBkt, dbChanIDInt)
                if err != nil {
                        return err
                }

                // Get the channel details bucket for the channel.
                chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
                if chanDetails == nil {
                        return fmt.Errorf("no channel details found for "+
                                "channel %s referenced by session %s", chanID,
                                id)
                }

                // If a closed height has been set, then the channel is closed.
                closedHeight := chanDetails.Get(cChanClosedHeight)
                if len(closedHeight) > 0 {
                        return nil
                }

                // Otherwise, the channel is not yet closed meaning that the
                // session is not yet closable. We break the ForEach by
                // returning an error to indicate this.
                return errSessionHasOpenChannels
        })
        if errors.Is(err, errSessionHasOpenChannels) {
                return false, nil
        } else if err != nil {
                return false, err
        }

        return true, nil
}

// CommitUpdate persists the CommittedUpdate provided in the slot for (session,
// seqNum). This allows the client to retransmit this update on startup.
func (c *ClientDB) CommitUpdate(id *SessionID,
        update *CommittedUpdate) (uint16, error) {

        var lastApplied uint16
        err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == 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 will be modified on disk
                // directly.
                session, err := getClientSessionBody(sessions, id[:])
                if err != nil {
                        return err
                }

                // Can't fail if the above didn't fail.
                sessionBkt := sessions.NestedReadWriteBucket(id[:])

                // Ensure the session commits sub-bucket is initialized.
                sessionCommits, err := sessionBkt.CreateBucketIfNotExists(
                        cSessionCommits,
                )
                if err != nil {
                        return err
                }

                var seqNumBuf [2]byte
                byteOrder.PutUint16(seqNumBuf[:], update.SeqNum)

                // Check to see if a committed update already exists for this
                // sequence number.
                committedUpdateBytes := sessionCommits.Get(seqNumBuf[:])
                if committedUpdateBytes != nil {
                        var dbUpdate CommittedUpdate
                        err := dbUpdate.Decode(
                                bytes.NewReader(committedUpdateBytes),
                        )
                        if err != nil {
                                return err
                        }

                        // If an existing committed update has a different hint,
                        // we'll reject this newer update.
                        if dbUpdate.Hint != update.Hint {
                                return ErrUpdateAlreadyCommitted
                        }

                        // Otherwise, capture the last applied value and
                        // succeed.
                        lastApplied = session.TowerLastApplied
                        return nil
                }

                // There's no committed update for this sequence number, ensure
                // that we are committing the next unallocated one.
                if update.SeqNum != session.SeqNum+1 {
                        return ErrCommitUnorderedUpdate
                }

                // Increment the session's sequence number and store the updated
                // client session.
                //
                // TODO(conner): split out seqnum and last applied own bucket to
                // eliminate serialization of full struct during CommitUpdate?
                // Can also read/write directly to byes [:2] without migration.
                session.SeqNum++
                err = putClientSessionBody(sessionBkt, session)
                if err != nil {
                        return err
                }

                // Encode and store the committed update in the sessionCommits
                // sub-bucket under the requested sequence number.
                var b bytes.Buffer
                err = update.Encode(&b)
                if err != nil {
                        return err
                }

                err = sessionCommits.Put(seqNumBuf[:], b.Bytes())
                if err != nil {
                        return err
                }

                // Update the channel's max commitment height if needed.
                err = maybeUpdateMaxCommitHeight(tx, update.BackupID)
                if err != nil {
                        return err
                }

                // Finally, capture the session's last applied value so it can
                // be sent in the next state update to the tower.
                lastApplied = session.TowerLastApplied

                return nil
        }, func() {
                lastApplied = 0
        })
        if err != nil {
                return 0, err
        }

        return lastApplied, nil
}

// AckUpdate persists an acknowledgment for a given (session, seqnum) pair. This
// removes the update from the set of committed updates, and validates the
// lastApplied value returned from the tower.
func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16,
        lastApplied uint16) error {

        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                chanDetailsBkt := tx.ReadWriteBucket(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
                // modified on disk directly.
                session, err := getClientSessionBody(sessions, id[:])
                if err != nil {
                        return err
                }

                // If the tower has acked a sequence number beyond our highest
                // sequence number, fail.
                if lastApplied > session.SeqNum {
                        return ErrUnallocatedLastApplied
                }

                // If the tower acked with a lower sequence number than it gave
                // us prior, fail.
                if lastApplied < session.TowerLastApplied {
                        return ErrLastAppliedReversion
                }

                // TODO(conner): split out seqnum and last applied own bucket to
                // eliminate serialization of full struct during AckUpdate?  Can
                // 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(sessionBkt, session)
                if err != nil {
                        return err
                }

                // If the commits sub-bucket doesn't exist, there can't possibly
                // be a corresponding committed update to remove.
                sessionCommits := sessionBkt.NestedReadWriteBucket(
                        cSessionCommits,
                )
                if sessionCommits == nil {
                        return ErrCommittedUpdateNotFound
                }

                var seqNumBuf [2]byte
                byteOrder.PutUint16(seqNumBuf[:], seqNum)

                // Assert that a committed update exists for this sequence
                // number.
                committedUpdateBytes := sessionCommits.Get(seqNumBuf[:])
                if committedUpdateBytes == nil {
                        return ErrCommittedUpdateNotFound
                }

                var committedUpdate CommittedUpdate
                err = committedUpdate.Decode(
                        bytes.NewReader(committedUpdateBytes),
                )
                if err != nil {
                        return err
                }

                // Remove the corresponding committed update.
                err = sessionCommits.Delete(seqNumBuf[:])
                if err != nil {
                        return err
                }

                dbSessionID, dbSessIDBytes, err := getDBSessionID(sessions, *id)
                if err != nil {
                        return err
                }

                chanID := committedUpdate.BackupID.ChanID
                height := committedUpdate.BackupID.CommitHeight

                // Get the DB representation of the channel-ID. There is a
                // chance that the channel corresponding to this update has been
                // closed and that the details for this channel no longer exist
                // in the tower client DB. In that case, we consider this a
                // rogue update and all we do is make sure to keep track of the
                // number of rogue updates for this session.
                _, dbChanIDBytes, err := getDBChanID(chanDetailsBkt, chanID)
                if errors.Is(err, ErrChannelNotRegistered) {
                        var (
                                count uint64
                                err   error
                        )

                        rogueCountBytes := sessionBkt.Get(
                                cSessionRogueUpdateCount,
                        )
                        if len(rogueCountBytes) != 0 {
                                count, err = readBigSize(rogueCountBytes)
                                if err != nil {
                                        return err
                                }
                        }

                        rogueCount := count + 1
                        countBytes, err := writeBigSize(rogueCount)
                        if err != nil {
                                return err
                        }

                        err = sessionBkt.Put(
                                cSessionRogueUpdateCount, countBytes,
                        )
                        if err != nil {
                                return err
                        }

                        // In the rare chance that this session only has rogue
                        // updates, we check here if the count is equal to the
                        // MaxUpdate of the session. If it is, then we mark the
                        // session as closable.
                        if rogueCount != uint64(session.Policy.MaxUpdates) {
                                return nil
                        }

                        // Before we mark the session as closable, we do a
                        // sanity check to ensure that this session has no
                        // acked-update index.
                        sessionAckRanges := sessionBkt.NestedReadBucket(
                                cSessionAckRangeIndex,
                        )
                        if sessionAckRanges != nil {
                                return fmt.Errorf("session(%s) has an "+
                                        "acked ranges index but has a rogue "+
                                        "count indicating saturation",
                                        session.ID)
                        }

                        closableSessBkt := tx.ReadWriteBucket(
                                cClosableSessionsBkt,
                        )
                        if closableSessBkt == nil {
                                return ErrUninitializedDB
                        }

                        var height [4]byte
                        byteOrder.PutUint32(height[:], 0)

                        return closableSessBkt.Put(dbSessIDBytes, height[:])
                } else if err != nil {
                        return err
                }

                // 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(
                        sessionBkt, dbChanIDBytes,
                )
                if err != nil {
                        return err
                }

                chanDetails := chanDetailsBkt.NestedReadWriteBucket(
                        committedUpdate.BackupID.ChanID[:],
                )
                if chanDetails == nil {
                        return ErrChannelNotRegistered
                }

                err = putChannelToSessionMapping(chanDetails, dbSessionID)
                if err != nil {
                        return err
                }

                // Get the range index for the given session-channel pair.
                index, err := c.getRangeIndex(tx, *id, chanID)
                if err != nil {
                        return err
                }

                return index.Add(height, rangesBkt)
        }, func() {})
}

// GetDBQueue returns a BackupID Queue instance under the given namespace.
func (c *ClientDB) GetDBQueue(namespace []byte) Queue[*BackupID] {
        return NewQueueDB(
                c.db, namespace, func() *BackupID {
                        return &BackupID{}
                }, func(tx kvdb.RwTx, item *BackupID) error {
                        return maybeUpdateMaxCommitHeight(tx, *item)
                },
        )
}

// TerminateSession sets the given session's status to CSessionTerminal meaning
// that it will not be usable again. An error will be returned if the given
// session still has un-acked updates that should be attended to.
func (c *ClientDB) TerminateSession(id SessionID) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                sessionsBkt := tx.ReadBucket(cSessionBkt)
                if sessionsBkt == nil {
                        return ErrUninitializedDB
                }

                chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
                if chanIDIndexBkt == nil {
                        return ErrUninitializedDB
                }

                // Collect any un-acked updates for this session.
                committedUpdateCount := make(map[SessionID]uint16)
                perCommittedUpdate := func(s *ClientSession,
                        _ *CommittedUpdate) {

                        committedUpdateCount[s.ID]++
                }

                session, err := c.getClientSession(
                        sessionsBkt, chanIDIndexBkt, id[:],
                        WithPerCommittedUpdate(perCommittedUpdate),
                )
                if err != nil {
                        return err
                }

                // If there are any un-acked updates for this session then
                // we don't allow the change of status as these updates must
                // first be dealt with somehow.
                if committedUpdateCount[id] > 0 {
                        return ErrSessionHasUnackedUpdates
                }

                return markSessionStatus(sessions, session, CSessionTerminal)
        }, func() {})
}

// DeleteCommittedUpdates deletes all the committed updates for the given
// session.
func (c *ClientDB) DeleteCommittedUpdates(id *SessionID) error {
        return kvdb.Update(c.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(cSessionBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                sessionBkt := sessions.NestedReadWriteBucket(id[:])
                if sessionBkt == nil {
                        return fmt.Errorf("session bucket %s not found",
                                id.String())
                }

                // If the commits sub-bucket doesn't exist, there can't possibly
                // be corresponding updates to remove.
                sessionCommits := sessionBkt.NestedReadWriteBucket(
                        cSessionCommits,
                )
                if sessionCommits == nil {
                        return nil
                }

                // errFoundUpdates is an error we will use to exit early from
                // the ForEach loop. The return of this error means that at
                // least one committed update exists.
                var errFoundUpdates = fmt.Errorf("found committed updates")
                err := sessionCommits.ForEach(func(k, v []byte) error {
                        return errFoundUpdates
                })
                switch {
                // If the errFoundUpdates signal error was returned then there
                // are some updates that need to be deleted.
                case errors.Is(err, errFoundUpdates):

                // If no error is returned then the ForEach call back was never
                // entered meaning that there are no un-acked committed updates.
                // So we can exit now as there is nothing left to do.
                case err == nil:
                        return nil

                // If an expected error is returned, return that error.
                default:
                        return err
                }

                session, err := getClientSessionBody(sessions, id[:])
                if err != nil {
                        return err
                }

                // Once we delete a committed update from the session, the
                // SeqNum of the session will be incorrect and so the session
                // should be marked as terminal.
                session.Status = CSessionTerminal
                err = putClientSessionBody(sessionBkt, session)
                if err != nil {
                        return err
                }

                // Delete all the committed updates in one go by deleting the
                // session commits bucket.
                return sessionBkt.DeleteNestedBucket(cSessionCommits)
        }, func() {})
}

// putChannelToSessionMapping adds the given session ID to a channel's
// cChanSessions bucket.
func putChannelToSessionMapping(chanDetails kvdb.RwBucket,
        dbSessID uint64) error {

        chanSessIDsBkt, err := chanDetails.CreateBucketIfNotExists(
                cChanSessions,
        )
        if err != nil {
                return err
        }

        b, err := writeBigSize(dbSessID)
        if err != nil {
                return err
        }

        return chanSessIDsBkt.Put(b, []byte{1})
}

// getClientSessionBody loads the body of a ClientSession from the sessions
// bucket corresponding to the serialized session id. This does not deserialize
// the CommittedUpdates, AckUpdates or the Tower associated with the session.
// If the caller requires this info, use getClientSession.
func getClientSessionBody(sessions kvdb.RBucket,
        idBytes []byte) (*ClientSession, error) {

        sessionBkt := sessions.NestedReadBucket(idBytes)
        if sessionBkt == nil {
                return nil, ErrClientSessionNotFound
        }

        // Should never have a sessionBkt without also having its body.
        sessionBody := sessionBkt.Get(cSessionBody)
        if sessionBody == nil {
                return nil, ErrCorruptClientSession
        }

        var session ClientSession
        copy(session.ID[:], idBytes)

        err := session.Decode(bytes.NewReader(sessionBody))
        if err != nil {
                return nil, err
        }

        return &session, nil
}

// ClientSessionFilterFn describes the signature of a callback function that can
// be used to filter the sessions that are returned in any of the DB methods
// that read sessions from the DB.
type ClientSessionFilterFn func(*ClientSession) bool

// ClientSessWithNumCommittedUpdatesFilterFn describes the signature of a
// callback function that can be used to filter out a session based on the
// contents of ClientSession along with the number of un-acked committed updates
// that the session has.
type ClientSessWithNumCommittedUpdatesFilterFn func(*ClientSession, uint16) bool

// 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)

// PerRogueUpdateCountCB describes the signature of a callback function that can
// be called for each session with the number of rogue updates that the session
// has.
type PerRogueUpdateCountCB func(*ClientSession, uint16)

// 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)

// PerCommittedUpdateCB describes the signature of a callback function that can
// be called for each of a session's committed updates (updates that the client
// has not yet received an ACK for).
type PerCommittedUpdateCB func(*ClientSession, *CommittedUpdate)

// ClientSessionListOption describes the signature of a functional option that
// can be used when listing client sessions in order to provide any extra
// instruction to the query.
type ClientSessionListOption func(cfg *ClientSessionListCfg)

// ClientSessionListCfg defines various query parameters that will be used when
// querying the DB for client sessions.
type ClientSessionListCfg struct {
        // 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

        // PerRogueUpdateCount will, if set, be called with the number of rogue
        // updates that the session has backed up.
        PerRogueUpdateCount PerRogueUpdateCountCB

        // 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.
        PerCommittedUpdate PerCommittedUpdateCB

        // PreEvaluateFilterFn will be run after loading a session from the DB
        // and _before_ any of the other call-back functions in
        // ClientSessionListCfg. Therefore, if a session fails this filter
        // function, then it will not be passed to any of the other call backs
        // and won't be included in the return list.
        PreEvaluateFilterFn ClientSessionFilterFn

        // PostEvaluateFilterFn will be run _after_ all the other call-back
        // functions in ClientSessionListCfg. If a session fails this filter
        // function then all it means is that it won't be included in the list
        // of sessions to return.
        PostEvaluateFilterFn ClientSessWithNumCommittedUpdatesFilterFn
}

// NewClientSessionCfg constructs a new ClientSessionListCfg.
func NewClientSessionCfg() *ClientSessionListCfg {
        return &ClientSessionListCfg{}
}

// 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.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
        }
}

// WithPerRogueUpdateCount constructs a functional option that will set a
// call-back function to be called with the number of rogue updates that the
// session has backed up.
func WithPerRogueUpdateCount(cb PerRogueUpdateCountCB) ClientSessionListOption {
        return func(cfg *ClientSessionListCfg) {
                cfg.PerRogueUpdateCount = cb
        }
}

// WithPerCommittedUpdate constructs a functional option that will set a
// call-back function to be called for each of a client's un-acked updates.
func WithPerCommittedUpdate(cb PerCommittedUpdateCB) ClientSessionListOption {
        return func(cfg *ClientSessionListCfg) {
                cfg.PerCommittedUpdate = cb
        }
}

// WithPreEvalFilterFn constructs a functional option that will set a call-back
// function that will be called immediately after loading a session. If the
// session fails this filter function, then it will not be passed to any of the
// other evaluation call-back functions.
func WithPreEvalFilterFn(fn ClientSessionFilterFn) ClientSessionListOption {
        return func(cfg *ClientSessionListCfg) {
                cfg.PreEvaluateFilterFn = fn
        }
}

// WithPostEvalFilterFn constructs a functional option that will set a call-back
// function that will be used to determine if a session should be included in
// the returned list. This differs from WithPreEvalFilterFn since that call-back
// is used to determine if the session should be evaluated at all (and thus
// run against the other ClientSessionListCfg call-backs) whereas the session
// will only reach the PostEvalFilterFn call-back once it has already been
// evaluated by all the other call-backs.
func WithPostEvalFilterFn(
        fn ClientSessWithNumCommittedUpdatesFilterFn) ClientSessionListOption {

        return func(cfg *ClientSessionListCfg) {
                cfg.PostEvaluateFilterFn = fn
        }
}

// 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 (c *ClientDB) getClientSession(sessionsBkt, chanIDIndexBkt kvdb.RBucket,
        idBytes []byte, opts ...ClientSessionListOption) (*ClientSession,
        error) {

        cfg := NewClientSessionCfg()
        for _, o := range opts {
                o(cfg)
        }

        session, err := getClientSessionBody(sessionsBkt, idBytes)
        if err != nil {
                return nil, err
        }

        if cfg.PreEvaluateFilterFn != nil && !cfg.PreEvaluateFilterFn(session) {
                return nil, ErrSessionFailedFilterFn
        }

        // Can't fail because client session body has already been read.
        sessionBkt := sessionsBkt.NestedReadBucket(idBytes)

        // Pass the session's committed (un-acked) updates through the call-back
        // if one is provided.
        numCommittedUpdates, err := filterClientSessionCommits(
                sessionBkt, session, cfg.PerCommittedUpdate,
        )
        if err != nil {
                return nil, err
        }

        // Pass the session's acked updates through the call-back if one is
        // provided.
        err = c.filterClientSessionAcks(
                sessionBkt, chanIDIndexBkt, session, cfg.PerMaxHeight,
                cfg.PerNumAckedUpdates, cfg.PerRogueUpdateCount,
        )
        if err != nil {
                return nil, err
        }

        if cfg.PostEvaluateFilterFn != nil &&
                !cfg.PostEvaluateFilterFn(session, numCommittedUpdates) {

                return nil, ErrSessionFailedFilterFn
        }

        return session, nil
}

// getClientSessionCommits retrieves all committed updates for the session
// identified by the serialized session id. If a PerCommittedUpdateCB is
// provided, then it will be called for each of the session's committed updates.
func getClientSessionCommits(sessionBkt kvdb.RBucket, s *ClientSession,
        cb PerCommittedUpdateCB) ([]CommittedUpdate, error) {

        // Initialize committedUpdates so that we can return an initialized map
        // if no committed updates exist.
        committedUpdates := make([]CommittedUpdate, 0)

        sessionCommits := sessionBkt.NestedReadBucket(cSessionCommits)
        if sessionCommits == nil {
                return committedUpdates, nil
        }

        err := sessionCommits.ForEach(func(k, v []byte) error {
                var committedUpdate CommittedUpdate
                err := committedUpdate.Decode(bytes.NewReader(v))
                if err != nil {
                        return err
                }
                committedUpdate.SeqNum = byteOrder.Uint16(k)

                committedUpdates = append(committedUpdates, committedUpdate)

                if cb != nil {
                        cb(s, &committedUpdate)
                }

                return nil
        })
        if err != nil {
                return nil, err
        }

        return committedUpdates, nil
}

// 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 (c *ClientDB) filterClientSessionAcks(sessionBkt,
        chanIDIndexBkt kvdb.RBucket, s *ClientSession, perMaxCb PerMaxHeightCB,
        perNumAckedUpdates PerNumAckedUpdatesCB,
        perRogueUpdateCount PerRogueUpdateCountCB) error {

        if perRogueUpdateCount != nil {
                var (
                        count uint64
                        err   error
                )
                rogueCountBytes := sessionBkt.Get(cSessionRogueUpdateCount)
                if len(rogueCountBytes) != 0 {
                        count, err = readBigSize(rogueCountBytes)
                        if err != nil {
                                return err
                        }
                }

                perRogueUpdateCount(s, uint16(count))
        }

        if perMaxCb == nil && perNumAckedUpdates == nil {
                return nil
        }

        sessionAcksRanges := sessionBkt.NestedReadBucket(cSessionAckRangeIndex)
        if sessionAcksRanges == nil {
                return nil
        }

        return sessionAcksRanges.ForEach(func(dbChanID, _ []byte) error {
                rangeBkt := sessionAcksRanges.NestedReadBucket(dbChanID)
                if rangeBkt == nil {
                        return nil
                }

                index, err := readRangeIndex(rangeBkt)
                if err != nil {
                        return err
                }

                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
        })
}

// filterClientSessionCommits retrieves all committed updates for the session
// identified by the serialized session id and passes them to the given
// PerCommittedUpdateCB callback.
func filterClientSessionCommits(sessionBkt kvdb.RBucket, s *ClientSession,
        cb PerCommittedUpdateCB) (uint16, error) {

        sessionCommits := sessionBkt.NestedReadBucket(cSessionCommits)
        if sessionCommits == nil {
                return 0, nil
        }

        var numUpdates uint16
        err := sessionCommits.ForEach(func(k, v []byte) error {
                numUpdates++

                if cb == nil {
                        return nil
                }

                var committedUpdate CommittedUpdate
                err := committedUpdate.Decode(bytes.NewReader(v))
                if err != nil {
                        return err
                }
                committedUpdate.SeqNum = byteOrder.Uint16(k)

                cb(s, &committedUpdate)

                return nil
        })
        if err != nil {
                return 0, err
        }

        return numUpdates, nil
}

// putClientSessionBody stores the body of the ClientSession (everything but the
// CommittedUpdates and AckedUpdates).
func putClientSessionBody(sessionBkt kvdb.RwBucket,
        session *ClientSession) error {

        var b bytes.Buffer
        err := session.Encode(&b)
        if err != nil {
                return err
        }

        return sessionBkt.Put(cSessionBody, b.Bytes())
}

// markSessionStatus updates the persisted state of the session to the new
// status.
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(sessionBkt, session)
}

// getChanSummary loads a ClientChanSummary for the passed chanID.
func getChanSummary(chanDetails kvdb.RBucket) (*ClientChanSummary, error) {
        chanSummaryBytes := chanDetails.Get(cChannelSummary)
        if chanSummaryBytes == nil {
                return nil, ErrChannelNotRegistered
        }

        var summary ClientChanSummary
        err := summary.Decode(bytes.NewReader(chanSummaryBytes))
        if err != nil {
                return nil, err
        }

        return &summary, nil
}

// putChanSummary stores a ClientChanSummary for the passed chanID.
func putChanSummary(chanDetails kvdb.RwBucket,
        summary *ClientChanSummary) error {

        var b bytes.Buffer
        err := summary.Encode(&b)
        if err != nil {
                return err
        }

        return chanDetails.Put(cChannelSummary, b.Bytes())
}

// getTower loads a Tower identified by its serialized tower id.
func getTower(towers kvdb.RBucket, id []byte) (*Tower, error) {
        towerBytes := towers.Get(id)
        if towerBytes == nil {
                return nil, ErrTowerNotFound
        }

        var tower Tower
        err := tower.Decode(bytes.NewReader(towerBytes))
        if err != nil {
                return nil, err
        }

        tower.ID = TowerIDFromBytes(id)

        return &tower, nil
}

// putTower stores a Tower identified by its serialized tower id.
func putTower(towers kvdb.RwBucket, tower *Tower) error {
        var b bytes.Buffer
        err := tower.Encode(&b)
        if err != nil {
                return err
        }

        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
}

// getDBSessionID returns the db-assigned session ID for the given real session
// ID. It returns both the uint64 and byte representation.
func getDBSessionID(sessionsBkt kvdb.RBucket, sessionID SessionID) (uint64,
        []byte, error) {

        sessionBkt := sessionsBkt.NestedReadBucket(sessionID[:])
        if sessionBkt == nil {
                return 0, nil, ErrClientSessionNotFound
        }

        idBytes := sessionBkt.Get(cSessionDBID)
        if len(idBytes) == 0 {
                return 0, nil, fmt.Errorf("no db-assigned ID found for "+
                        "session ID %s", sessionID)
        }

        id, err := readBigSize(idBytes)
        if err != nil {
                return 0, nil, err
        }

        return id, idBytes, nil
}

// maybeUpdateMaxCommitHeight updates the given channel details bucket with the
// given height if it is larger than the current max height stored for the
// channel.
func maybeUpdateMaxCommitHeight(tx kvdb.RwTx, backupID BackupID) error {
        chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
        if chanDetailsBkt == nil {
                return ErrUninitializedDB
        }

        // If an entry for this channel does not exist in the channel details
        // bucket then we exit here as this means that the channel has been
        // closed.
        chanDetails := chanDetailsBkt.NestedReadWriteBucket(backupID.ChanID[:])
        if chanDetails == nil {
                return nil
        }

        putHeight := func() error {
                b, err := writeBigSize(backupID.CommitHeight)
                if err != nil {
                        return err
                }

                return chanDetails.Put(
                        cChanMaxCommitmentHeight, b,
                )
        }

        // Get current height.
        heightBytes := chanDetails.Get(cChanMaxCommitmentHeight)

        // The height might have not been set yet, in which case
        // we can just write the new height.
        if len(heightBytes) == 0 {
                return putHeight()
        }

        // Otherwise, read in the current max commitment height for the channel.
        currentHeight, err := readBigSize(heightBytes)
        if err != nil {
                return err
        }

        // If the new height is not larger than the current persisted height,
        // then there is nothing left for us to do.
        if backupID.CommitHeight <= currentHeight {
                return nil
        }

        return putHeight()
}

func getRealSessionID(sessIDIndexBkt kvdb.RBucket, dbID uint64) (*SessionID,
        error) {

        dbIDBytes, err := writeBigSize(dbID)
        if err != nil {
                return nil, err
        }

        sessIDBytes := sessIDIndexBkt.Get(dbIDBytes)
        if len(sessIDBytes) != SessionIDSize {
                return nil, fmt.Errorf("session ID not found")
        }

        var sessID SessionID
        copy(sessID[:], sessIDBytes)

        return &sessID, nil
}

func getRealChannelID(chanIDIndexBkt kvdb.RBucket,
        dbID uint64) (*lnwire.ChannelID, error) {

        dbIDBytes, err := writeBigSize(dbID)
        if err != nil {
                return nil, err
        }

        chanIDBytes := chanIDIndexBkt.Get(dbIDBytes)
        if len(chanIDBytes) != 32 { //nolint:gomnd
                return nil, fmt.Errorf("channel ID not found")
        }

        var chanIDS lnwire.ChannelID
        copy(chanIDS[:], chanIDBytes)

        return &chanIDS, 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
}

lightningnetwork / lnd / 12250203495

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

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