13566028875

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

Build # 13566028875

Build Type

Pull #9555

github

Committed by

ellemouton

Commit Message

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

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

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

Run Details

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

27464 existing lines in 436 files now uncovered.

101095 of 204664 relevant lines covered (49.4%)

1.54 hits per line

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

64.46

/watchtower/wtdb/tower_db.go

package wtdb

import (
        "bytes"
        "errors"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/watchtower/blob"
)

var (
        // sessionsBkt is a bucket containing all negotiated client sessions.
        //  session id -> session
        sessionsBkt = []byte("sessions-bucket")

        // updatesBkt is a bucket containing all state updates sent by clients.
        // The updates are further bucketed by session id to prevent clients
        // from overwrite each other.
        //   hint => session id -> update
        updatesBkt = []byte("updates-bucket")

        // updateIndexBkt is a bucket that indexes all state updates by their
        // overarching session id. This allows for efficient lookup of updates
        // by their session id, which is currently used to aide deletion
        // performance.
        //  session id => hint1 -> []byte{}
        //             => hint2 -> []byte{}
        updateIndexBkt = []byte("update-index-bucket")

        // lookoutTipBkt is a bucket containing the last block epoch processed
        // by the lookout subsystem. It has one key, lookoutTipKey.
        //   lookoutTipKey -> block epoch
        lookoutTipBkt = []byte("lookout-tip-bucket")

        // lookoutTipKey is a static key used to retrieve lookout tip's block
        // epoch from the lookoutTipBkt.
        lookoutTipKey = []byte("lookout-tip")

        // ErrNoSessionHintIndex signals that an active session does not have an
        // initialized index for tracking its own state updates.
        ErrNoSessionHintIndex = errors.New("session hint index missing")

        // ErrInvalidBlobSize indicates that the encrypted blob provided by the
        // client is not valid according to the blob type of the session.
        ErrInvalidBlobSize = errors.New("invalid blob size")
)

// TowerDB is single database providing a persistent storage engine for the
// wtserver and lookout subsystems.
type TowerDB struct {
        db kvdb.Backend
}

// OpenTowerDB opens the tower 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 OpenTowerDB(db kvdb.Backend) (*TowerDB, error) {
        firstInit, err := isFirstInit(db)
        if err != nil {
                return nil, err
        }

        towerDB := &TowerDB{
                db: db,
        }

        err = initOrSyncVersions(towerDB, firstInit, towerDBVersions)
        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(towerDB.db, initTowerDBBuckets, func() {})
        if err != nil {
                db.Close()
                return nil, err
        }

        return towerDB, nil
}

// initTowerDBBuckets creates all top-level buckets required to handle database
// operations required by the latest version.
func initTowerDBBuckets(tx kvdb.RwTx) error {
        buckets := [][]byte{
                sessionsBkt,
                updateIndexBkt,
                updatesBkt,
                lookoutTipBkt,
        }

        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 (t *TowerDB) bdb() kvdb.Backend {
        return t.db
}

// Version returns the database's current version number.
//
// NOTE: Part of the versionedDB interface.
func (t *TowerDB) Version() (uint32, error) {
        var version uint32
        err := kvdb.View(t.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 (t *TowerDB) Close() error {
        return t.db.Close()
}

// GetSessionInfo retrieves the session for the passed session id. An error is
// returned if the session could not be found.
func (t *TowerDB) GetSessionInfo(id *SessionID) (*SessionInfo, error) {
        var session *SessionInfo
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                var err error
                session, err = getSession(sessions, id[:])
                return err
        }, func() {
                session = nil
        })
        if err != nil {
                return nil, err
        }

        return session, nil
}

// InsertSessionInfo records a negotiated session in the tower database. An
// error is returned if the session already exists.
func (t *TowerDB) InsertSessionInfo(session *SessionInfo) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                dbSession, err := getSession(sessions, session.ID[:])
                switch {
                case err == ErrSessionNotFound:
                        // proceed.

                case err != nil:
                        return err

                case dbSession.LastApplied > 0:
                        return ErrSessionAlreadyExists
                }

                // Perform a quick sanity check on the session policy before
                // accepting.
                if err := session.Policy.Validate(); err != nil {
                        return err
                }

                err = putSession(sessions, session)
                if err != nil {
                        return err
                }

                // Initialize the session-hint index which will be used to track
                // all updates added for this session. Upon deletion, we will
                // consult the index to determine exactly which updates should
                // be deleted without needing to iterate over the entire
                // database.
                return touchSessionHintBkt(updateIndex, &session.ID)
        }, func() {})
}

// InsertStateUpdate stores an update sent by the client after validating that
// the update is well-formed in the context of other updates sent for the same
// session. This include verifying that the sequence number is incremented
// properly and the last applied values echoed by the client are sane.
func (t *TowerDB) InsertStateUpdate(update *SessionStateUpdate) (uint16, error) {
        var lastApplied uint16
        err := kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadWriteBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                // Fetch the session corresponding to the update's session id.
                // This will be used to validate that the update's sequence
                // number and last applied values are sane.
                session, err := getSession(sessions, update.ID[:])
                if err != nil {
                        return err
                }

                commitType, err := session.Policy.BlobType.CommitmentType(nil)
                if err != nil {
                        return err
                }

                kit, err := commitType.EmptyJusticeKit()
                if err != nil {
                        return err
                }

                // Assert that the blob is the correct size for the session's
                // blob type.
                expBlobSize := blob.Size(kit)
                if len(update.EncryptedBlob) != expBlobSize {
                        return ErrInvalidBlobSize
                }

                // Validate the update against the current state of the session.
                err = session.AcceptUpdateSequence(
                        update.SeqNum, update.LastApplied,
                )
                if err != nil {
                        return err
                }

                // Validation succeeded, therefore the update is committed and
                // the session's last applied value is equal to the update's
                // sequence number.
                lastApplied = session.LastApplied

                // Store the updated session to persist the updated last applied
                // values.
                err = putSession(sessions, session)
                if err != nil {
                        return err
                }

                // Create or load the hint bucket for this state update's hint
                // and write the given update.
                hints, err := updates.CreateBucketIfNotExists(update.Hint[:])
                if err != nil {
                        return err
                }

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

                err = hints.Put(update.ID[:], b.Bytes())
                if err != nil {
                        return err
                }

                // Finally, create an entry in the update index to track this
                // hint under its session id. This will allow us to delete the
                // entries efficiently if the session is ever removed.
                return putHintForSession(updateIndex, &update.ID, update.Hint)
        }, func() {
                lastApplied = 0
        })
        if err != nil {
                return 0, err
        }

        return lastApplied, nil
}

// DeleteSession removes all data associated with a particular session id from
// the tower's database.
func (t *TowerDB) DeleteSession(target SessionID) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadWriteBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                // Fail if the session doesn't exit.
                _, err := getSession(sessions, target[:])
                if err != nil {
                        return err
                }

                // Remove the target session.
                err = sessions.Delete(target[:])
                if err != nil {
                        return err
                }

                // Next, check the update index for any hints that were added
                // under this session.
                hints, err := getHintsForSession(updateIndex, &target)
                if err != nil {
                        return err
                }

                for _, hint := range hints {
                        // Remove the state updates for any blobs stored under
                        // the target session identifier.
                        updatesForHint := updates.NestedReadWriteBucket(hint[:])
                        if updatesForHint == nil {
                                continue
                        }

                        update := updatesForHint.Get(target[:])
                        if update == nil {
                                continue
                        }

                        err := updatesForHint.Delete(target[:])
                        if err != nil {
                                return err
                        }

                        // If this was the last state update, we can also remove
                        // the hint that would map to an empty set.
                        err = isBucketEmpty(updatesForHint)
                        switch {

                        // Other updates exist for this hint, keep the bucket.
                        case err == errBucketNotEmpty:
                                continue

                        // Unexpected error.
                        case err != nil:
                                return err

                        // No more updates for this hint, prune hint bucket.
                        default:
                                err = updates.DeleteNestedBucket(hint[:])
                                if err != nil {
                                        return err
                                }
                        }
                }

                // Finally, remove this session from the update index, which
                // also removes any of the indexed hints beneath it.
                return removeSessionHintBkt(updateIndex, &target)
        }, func() {})
}

// QueryMatches searches against all known state updates for any that match the
// passed breachHints. More than one Match will be returned for a given hint if
// they exist in the database.
func (t *TowerDB) QueryMatches(breachHints []blob.BreachHint) ([]Match, error) {
        var matches []Match
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                // Iterate through the target breach hints, appending any
                // matching updates to the set of matches.
                for _, hint := range breachHints {
                        // If a bucket does not exist for this hint, no matches
                        // are known.
                        updatesForHint := updates.NestedReadBucket(hint[:])
                        if updatesForHint == nil {
                                continue
                        }

                        // Otherwise, iterate through all (session id, update)
                        // pairs, creating a Match for each.
                        err := updatesForHint.ForEach(func(k, v []byte) error {
                                // Load the session via the session id for this
                                // update. The session info contains further
                                // instructions for how to process the state
                                // update.
                                session, err := getSession(sessions, k)
                                switch {
                                case err == ErrSessionNotFound:
                                        log.Warnf("Missing session=%x for "+
                                                "matched state update hint=%x",
                                                k, hint)
                                        return nil

                                case err != nil:
                                        return err
                                }

                                // Decode the state update containing the
                                // encrypted blob.
                                update := &SessionStateUpdate{}
                                err = update.Decode(bytes.NewReader(v))
                                if err != nil {
                                        return err
                                }

                                var id SessionID
                                copy(id[:], k)

                                // Construct the final match using the found
                                // update and its session info.
                                match := Match{
                                        ID:            id,
                                        SeqNum:        update.SeqNum,
                                        Hint:          hint,
                                        EncryptedBlob: update.EncryptedBlob,
                                        SessionInfo:   session,
                                }

                                matches = append(matches, match)

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

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

        return matches, nil
}

// SetLookoutTip stores the provided epoch as the latest lookout tip epoch in
// the tower database.
func (t *TowerDB) SetLookoutTip(epoch *chainntnfs.BlockEpoch) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                lookoutTip := tx.ReadWriteBucket(lookoutTipBkt)
                if lookoutTip == nil {
                        return ErrUninitializedDB
                }

                return putLookoutEpoch(lookoutTip, epoch)
        }, func() {})
}

// GetLookoutTip retrieves the current lookout tip block epoch from the tower
// database.
func (t *TowerDB) GetLookoutTip() (*chainntnfs.BlockEpoch, error) {
        var epoch *chainntnfs.BlockEpoch
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                lookoutTip := tx.ReadBucket(lookoutTipBkt)
                if lookoutTip == nil {
                        return ErrUninitializedDB
                }

                epoch = getLookoutEpoch(lookoutTip)

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

        return epoch, nil
}

// getSession retrieves the session info from the sessions bucket identified by
// its session id. An error is returned if the session is not found or a
// deserialization error occurs.
func getSession(sessions kvdb.RBucket, id []byte) (*SessionInfo, error) {
        sessionBytes := sessions.Get(id)
        if sessionBytes == nil {
                return nil, ErrSessionNotFound
        }

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

        return &session, nil
}

// putSession stores the session info in the sessions bucket identified by its
// session id. An error is returned if a serialization error occurs.
func putSession(sessions kvdb.RwBucket, session *SessionInfo) error {
        var b bytes.Buffer
        err := session.Encode(&b)
        if err != nil {
                return err
        }

        return sessions.Put(session.ID[:], b.Bytes())
}

// touchSessionHintBkt initializes the session-hint bucket for a particular
// session id. This ensures that future calls to getHintsForSession or
// putHintForSession can rely on the bucket already being created, and fail if
// index has not been initialized as this points to improper usage.
func touchSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {
        _, err := updateIndex.CreateBucketIfNotExists(id[:])
        return err
}

// removeSessionHintBkt prunes the session-hint bucket for the given session id
// and all of the hints contained inside. This should be used to clean up the
// index upon session deletion.
func removeSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {
        return updateIndex.DeleteNestedBucket(id[:])
}

// getHintsForSession returns all known hints belonging to the given session id.
// If the index for the session has not been initialized, this method returns
// ErrNoSessionHintIndex.
func getHintsForSession(updateIndex kvdb.RBucket,
        id *SessionID) ([]blob.BreachHint, error) {

        sessionHints := updateIndex.NestedReadBucket(id[:])
        if sessionHints == nil {
                return nil, ErrNoSessionHintIndex
        }

        var hints []blob.BreachHint
        err := sessionHints.ForEach(func(k, _ []byte) error {
                if len(k) != blob.BreachHintSize {
                        return nil
                }

                var hint blob.BreachHint
                copy(hint[:], k)
                hints = append(hints, hint)
                return nil
        })
        if err != nil {
                return nil, err
        }

        return hints, nil
}

// putHintForSession inserts a record into the update index for a given
// (session, hint) pair. The hints are coalesced under a bucket for the target
// session id, and used to perform efficient removal of updates. If the index
// for the session has not been initialized, this method returns
// ErrNoSessionHintIndex.
func putHintForSession(updateIndex kvdb.RwBucket, id *SessionID,
        hint blob.BreachHint) error {

        sessionHints := updateIndex.NestedReadWriteBucket(id[:])
        if sessionHints == nil {
                return ErrNoSessionHintIndex
        }

        return sessionHints.Put(hint[:], []byte{})
}

// putLookoutEpoch stores the given lookout tip block epoch in provided bucket.
func putLookoutEpoch(bkt kvdb.RwBucket, epoch *chainntnfs.BlockEpoch) error {
        epochBytes := make([]byte, 36)
        copy(epochBytes, epoch.Hash[:])
        byteOrder.PutUint32(epochBytes[32:], uint32(epoch.Height))

        return bkt.Put(lookoutTipKey, epochBytes)
}

// getLookoutEpoch retrieves the lookout tip block epoch from the given bucket.
// A nil epoch is returned if no update exists.
func getLookoutEpoch(bkt kvdb.RBucket) *chainntnfs.BlockEpoch {
        epochBytes := bkt.Get(lookoutTipKey)
        if len(epochBytes) != 36 {
                return nil
        }

        var hash chainhash.Hash
        copy(hash[:], epochBytes[:32])
        height := byteOrder.Uint32(epochBytes[32:])

        return &chainntnfs.BlockEpoch{
                Hash:   &hash,
                Height: int32(height),
        }
}

// errBucketNotEmpty is a helper error returned when testing whether a bucket is
// empty or not.
var errBucketNotEmpty = errors.New("bucket not empty")

// isBucketEmpty returns errBucketNotEmpty if the bucket is not empty.
func isBucketEmpty(bkt kvdb.RBucket) error {
        return bkt.ForEach(func(_, _ []byte) error {
                return errBucketNotEmpty
        })
}

1	package wtdb
2
3	import (
4	"bytes"
5	"errors"
6
7	"github.com/btcsuite/btcd/chaincfg/chainhash"
8	"github.com/lightningnetwork/lnd/chainntnfs"
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/watchtower/blob"
11	)
12
13	var (
14	// sessionsBkt is a bucket containing all negotiated client sessions.
15	// session id -> session
16	sessionsBkt = []byte("sessions-bucket")
17
18	// updatesBkt is a bucket containing all state updates sent by clients.
19	// The updates are further bucketed by session id to prevent clients
20	// from overwrite each other.
21	// hint => session id -> update
22	updatesBkt = []byte("updates-bucket")
23
24	// updateIndexBkt is a bucket that indexes all state updates by their
25	// overarching session id. This allows for efficient lookup of updates
26	// by their session id, which is currently used to aide deletion
27	// performance.
28	// session id => hint1 -> []byte{}
29	// => hint2 -> []byte{}
30	updateIndexBkt = []byte("update-index-bucket")
31
32	// lookoutTipBkt is a bucket containing the last block epoch processed
33	// by the lookout subsystem. It has one key, lookoutTipKey.
34	// lookoutTipKey -> block epoch
35	lookoutTipBkt = []byte("lookout-tip-bucket")
36
37	// lookoutTipKey is a static key used to retrieve lookout tip's block
38	// epoch from the lookoutTipBkt.
39	lookoutTipKey = []byte("lookout-tip")
40
41	// ErrNoSessionHintIndex signals that an active session does not have an
42	// initialized index for tracking its own state updates.
43	ErrNoSessionHintIndex = errors.New("session hint index missing")
44
45	// ErrInvalidBlobSize indicates that the encrypted blob provided by the
46	// client is not valid according to the blob type of the session.
47	ErrInvalidBlobSize = errors.New("invalid blob size")
48	)
49
50	// TowerDB is single database providing a persistent storage engine for the
51	// wtserver and lookout subsystems.
52	type TowerDB struct {
53	db kvdb.Backend
54	}
55
56	// OpenTowerDB opens the tower database given the path to the database's
57	// directory. If no such database exists, this method will initialize a fresh
58	// one using the latest version number and bucket structure. If a database
59	// exists but has a lower version number than the current version, any necessary
60	// migrations will be applied before returning. Any attempt to open a database
61	// with a version number higher that the latest version will fail to prevent
62	// accidental reversion.
63	func OpenTowerDB(db kvdb.Backend) (*TowerDB, error) {	3✔
64	firstInit, err := isFirstInit(db)	3✔
65	if err != nil {	3✔
66	return nil, err	×
67	}	×
68
69	towerDB := &TowerDB{	3✔
70	db: db,	3✔
71	}	3✔
72		3✔
73	err = initOrSyncVersions(towerDB, firstInit, towerDBVersions)	3✔
74	if err != nil {	3✔
75	db.Close()	×
76	return nil, err	×
77	}	×
78
79	// Now that the database version fully consistent with our latest known
80	// version, ensure that all top-level buckets known to this version are
81	// initialized. This allows us to assume their presence throughout all
82	// operations. If an known top-level bucket is expected to exist but is
83	// missing, this will trigger a ErrUninitializedDB error.
84	err = kvdb.Update(towerDB.db, initTowerDBBuckets, func() {})	6✔
85	if err != nil {	3✔
86	db.Close()	×
87	return nil, err	×
88	}	×
89
90	return towerDB, nil	3✔
91	}
92
93	// initTowerDBBuckets creates all top-level buckets required to handle database
94	// operations required by the latest version.
95	func initTowerDBBuckets(tx kvdb.RwTx) error {	3✔
96	buckets := [][]byte{	3✔
97	sessionsBkt,	3✔
98	updateIndexBkt,	3✔
99	updatesBkt,	3✔
100	lookoutTipBkt,	3✔
101	}	3✔
102		3✔
103	for _, bucket := range buckets {	6✔
104	_, err := tx.CreateTopLevelBucket(bucket)	3✔
105	if err != nil {	3✔
106	return err	×
107	}	×
108	}
109
110	return nil	3✔
111	}
112
113	// bdb returns the backing bbolt.DB instance.
114	//
115	// NOTE: Part of the versionedDB interface.
116	func (t *TowerDB) bdb() kvdb.Backend {	3✔
117	return t.db	3✔
118	}	3✔
119
120	// Version returns the database's current version number.
121	//
122	// NOTE: Part of the versionedDB interface.
UNCOV 123	func (t *TowerDB) Version() (uint32, error) {	×
UNCOV 124	var version uint32	×
UNCOV 125	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	×
UNCOV 126	var err error	×
UNCOV 127	version, err = getDBVersion(tx)	×
UNCOV 128	return err	×
UNCOV 129	}, func() {	×
UNCOV 130	version = 0	×
UNCOV 131	})	×
UNCOV 132	if err != nil {	×
133	return 0, err	×
134	}	×
135
UNCOV 136	return version, nil	×
137	}
138
139	// Close closes the underlying database.
UNCOV 140	func (t *TowerDB) Close() error {	×
UNCOV 141	return t.db.Close()	×
UNCOV 142	}	×
143
144	// GetSessionInfo retrieves the session for the passed session id. An error is
145	// returned if the session could not be found.
146	func (t TowerDB) GetSessionInfo(id SessionID) (*SessionInfo, error) {	3✔
147	var session *SessionInfo	3✔
148	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	6✔
149	sessions := tx.ReadBucket(sessionsBkt)	3✔
150	if sessions == nil {	3✔
151	return ErrUninitializedDB	×
152	}	×
153
154	var err error	3✔
155	session, err = getSession(sessions, id[:])	3✔
156	return err	3✔
157	}, func() {	3✔
158	session = nil	3✔
159	})	3✔
160	if err != nil {	6✔
161	return nil, err	3✔
162	}	3✔
163
UNCOV 164	return session, nil	×
165	}
166
167	// InsertSessionInfo records a negotiated session in the tower database. An
168	// error is returned if the session already exists.
169	func (t TowerDB) InsertSessionInfo(session SessionInfo) error {	3✔
170	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	6✔
171	sessions := tx.ReadWriteBucket(sessionsBkt)	3✔
172	if sessions == nil {	3✔
173	return ErrUninitializedDB	×
174	}	×
175
176	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	3✔
177	if updateIndex == nil {	3✔
178	return ErrUninitializedDB	×
179	}	×
180
181	dbSession, err := getSession(sessions, session.ID[:])	3✔
182	switch {	3✔
183	case err == ErrSessionNotFound:	3✔
184	// proceed.
185
186	case err != nil:	×
187	return err	×
188
UNCOV 189	case dbSession.LastApplied > 0:	×
UNCOV 190	return ErrSessionAlreadyExists	×
191	}
192
193	// Perform a quick sanity check on the session policy before
194	// accepting.
195	if err := session.Policy.Validate(); err != nil {	3✔
UNCOV 196	return err	×
UNCOV 197	}	×
198
199	err = putSession(sessions, session)	3✔
200	if err != nil {	3✔
201	return err	×
202	}	×
203
204	// Initialize the session-hint index which will be used to track
205	// all updates added for this session. Upon deletion, we will
206	// consult the index to determine exactly which updates should
207	// be deleted without needing to iterate over the entire
208	// database.
209	return touchSessionHintBkt(updateIndex, &session.ID)	3✔
210	}, func() {})	3✔
211	}
212
213	// InsertStateUpdate stores an update sent by the client after validating that
214	// the update is well-formed in the context of other updates sent for the same
215	// session. This include verifying that the sequence number is incremented
216	// properly and the last applied values echoed by the client are sane.
217	func (t TowerDB) InsertStateUpdate(update SessionStateUpdate) (uint16, error) {	3✔
218	var lastApplied uint16	3✔
219	err := kvdb.Update(t.db, func(tx kvdb.RwTx) error {	6✔
220	sessions := tx.ReadWriteBucket(sessionsBkt)	3✔
221	if sessions == nil {	3✔
222	return ErrUninitializedDB	×
223	}	×
224
225	updates := tx.ReadWriteBucket(updatesBkt)	3✔
226	if updates == nil {	3✔
227	return ErrUninitializedDB	×
228	}	×
229
230	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	3✔
231	if updateIndex == nil {	3✔
232	return ErrUninitializedDB	×
233	}	×
234
235	// Fetch the session corresponding to the update's session id.
236	// This will be used to validate that the update's sequence
237	// number and last applied values are sane.
238	session, err := getSession(sessions, update.ID[:])	3✔
239	if err != nil {	3✔
UNCOV 240	return err	×
UNCOV 241	}	×
242
243	commitType, err := session.Policy.BlobType.CommitmentType(nil)	3✔
244	if err != nil {	3✔
245	return err	×
246	}	×
247
248	kit, err := commitType.EmptyJusticeKit()	3✔
249	if err != nil {	3✔
250	return err	×
251	}	×
252
253	// Assert that the blob is the correct size for the session's
254	// blob type.
255	expBlobSize := blob.Size(kit)	3✔
256	if len(update.EncryptedBlob) != expBlobSize {	3✔
UNCOV 257	return ErrInvalidBlobSize	×
UNCOV 258	}	×
259
260	// Validate the update against the current state of the session.
261	err = session.AcceptUpdateSequence(	3✔
262	update.SeqNum, update.LastApplied,	3✔
263	)	3✔
264	if err != nil {	3✔
UNCOV 265	return err	×
UNCOV 266	}	×
267
268	// Validation succeeded, therefore the update is committed and
269	// the session's last applied value is equal to the update's
270	// sequence number.
271	lastApplied = session.LastApplied	3✔
272		3✔
273	// Store the updated session to persist the updated last applied	3✔
274	// values.	3✔
275	err = putSession(sessions, session)	3✔
276	if err != nil {	3✔
277	return err	×
278	}	×
279
280	// Create or load the hint bucket for this state update's hint
281	// and write the given update.
282	hints, err := updates.CreateBucketIfNotExists(update.Hint[:])	3✔
283	if err != nil {	3✔
284	return err	×
285	}	×
286
287	var b bytes.Buffer	3✔
288	err = update.Encode(&b)	3✔
289	if err != nil {	3✔
290	return err	×
291	}	×
292
293	err = hints.Put(update.ID[:], b.Bytes())	3✔
294	if err != nil {	3✔
295	return err	×
296	}	×
297
298	// Finally, create an entry in the update index to track this
299	// hint under its session id. This will allow us to delete the
300	// entries efficiently if the session is ever removed.
301	return putHintForSession(updateIndex, &update.ID, update.Hint)	3✔
302	}, func() {	3✔
303	lastApplied = 0	3✔
304	})	3✔
305	if err != nil {	3✔
UNCOV 306	return 0, err	×
UNCOV 307	}	×
308
309	return lastApplied, nil	3✔
310	}
311
312	// DeleteSession removes all data associated with a particular session id from
313	// the tower's database.
314	func (t *TowerDB) DeleteSession(target SessionID) error {	3✔
315	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	6✔
316	sessions := tx.ReadWriteBucket(sessionsBkt)	3✔
317	if sessions == nil {	3✔
318	return ErrUninitializedDB	×
319	}	×
320
321	updates := tx.ReadWriteBucket(updatesBkt)	3✔
322	if updates == nil {	3✔
323	return ErrUninitializedDB	×
324	}	×
325
326	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	3✔
327	if updateIndex == nil {	3✔
328	return ErrUninitializedDB	×
329	}	×
330
331	// Fail if the session doesn't exit.
332	_, err := getSession(sessions, target[:])	3✔
333	if err != nil {	3✔
UNCOV 334	return err	×
UNCOV 335	}	×
336
337	// Remove the target session.
338	err = sessions.Delete(target[:])	3✔
339	if err != nil {	3✔
340	return err	×
341	}	×
342
343	// Next, check the update index for any hints that were added
344	// under this session.
345	hints, err := getHintsForSession(updateIndex, &target)	3✔
346	if err != nil {	3✔
347	return err	×
348	}	×
349
350	for _, hint := range hints {	6✔
351	// Remove the state updates for any blobs stored under	3✔
352	// the target session identifier.	3✔
353	updatesForHint := updates.NestedReadWriteBucket(hint[:])	3✔
354	if updatesForHint == nil {	3✔
355	continue	×
356	}
357
358	update := updatesForHint.Get(target[:])	3✔
359	if update == nil {	3✔
360	continue	×
361	}
362
363	err := updatesForHint.Delete(target[:])	3✔
364	if err != nil {	3✔
365	return err	×
366	}	×
367
368	// If this was the last state update, we can also remove
369	// the hint that would map to an empty set.
370	err = isBucketEmpty(updatesForHint)	3✔
371	switch {	3✔
372
373	// Other updates exist for this hint, keep the bucket.
UNCOV 374	case err == errBucketNotEmpty:	×
UNCOV 375	continue	×
376
377	// Unexpected error.
378	case err != nil:	×
379	return err	×
380
381	// No more updates for this hint, prune hint bucket.
382	default:	3✔
383	err = updates.DeleteNestedBucket(hint[:])	3✔
384	if err != nil {	3✔
385	return err	×
386	}	×
387	}
388	}
389
390	// Finally, remove this session from the update index, which
391	// also removes any of the indexed hints beneath it.
392	return removeSessionHintBkt(updateIndex, &target)	3✔
393	}, func() {})	3✔
394	}
395
396	// QueryMatches searches against all known state updates for any that match the
397	// passed breachHints. More than one Match will be returned for a given hint if
398	// they exist in the database.
399	func (t *TowerDB) QueryMatches(breachHints []blob.BreachHint) ([]Match, error) {	3✔
400	var matches []Match	3✔
401	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	6✔
402	sessions := tx.ReadBucket(sessionsBkt)	3✔
403	if sessions == nil {	3✔
404	return ErrUninitializedDB	×
405	}	×
406
407	updates := tx.ReadBucket(updatesBkt)	3✔
408	if updates == nil {	3✔
409	return ErrUninitializedDB	×
410	}	×
411
412	// Iterate through the target breach hints, appending any
413	// matching updates to the set of matches.
414	for _, hint := range breachHints {	6✔
415	// If a bucket does not exist for this hint, no matches	3✔
416	// are known.	3✔
417	updatesForHint := updates.NestedReadBucket(hint[:])	3✔
418	if updatesForHint == nil {	6✔
419	continue	3✔
420	}
421
422	// Otherwise, iterate through all (session id, update)
423	// pairs, creating a Match for each.
424	err := updatesForHint.ForEach(func(k, v []byte) error {	6✔
425	// Load the session via the session id for this	3✔
426	// update. The session info contains further	3✔
427	// instructions for how to process the state	3✔
428	// update.	3✔
429	session, err := getSession(sessions, k)	3✔
430	switch {	3✔
431	case err == ErrSessionNotFound:	×
432	log.Warnf("Missing session=%x for "+	×
433	"matched state update hint=%x",	×
434	k, hint)	×
435	return nil	×
436
437	case err != nil:	×
438	return err	×
439	}
440
441	// Decode the state update containing the
442	// encrypted blob.
443	update := &SessionStateUpdate{}	3✔
444	err = update.Decode(bytes.NewReader(v))	3✔
445	if err != nil {	3✔
446	return err	×
447	}	×
448
449	var id SessionID	3✔
450	copy(id[:], k)	3✔
451		3✔
452	// Construct the final match using the found	3✔
453	// update and its session info.	3✔
454	match := Match{	3✔
455	ID: id,	3✔
456	SeqNum: update.SeqNum,	3✔
457	Hint: hint,	3✔
458	EncryptedBlob: update.EncryptedBlob,	3✔
459	SessionInfo: session,	3✔
460	}	3✔
461		3✔
462	matches = append(matches, match)	3✔
463		3✔
464	return nil	3✔
465	})
466	if err != nil {	3✔
467	return err	×
468	}	×
469	}
470
471	return nil	3✔
472	}, func() {	3✔
473	matches = nil	3✔
474	})	3✔
475	if err != nil {	3✔
476	return nil, err	×
477	}	×
478
479	return matches, nil	3✔
480	}
481
482	// SetLookoutTip stores the provided epoch as the latest lookout tip epoch in
483	// the tower database.
484	func (t TowerDB) SetLookoutTip(epoch chainntnfs.BlockEpoch) error {	3✔
485	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	6✔
486	lookoutTip := tx.ReadWriteBucket(lookoutTipBkt)	3✔
487	if lookoutTip == nil {	3✔
488	return ErrUninitializedDB	×
489	}	×
490
491	return putLookoutEpoch(lookoutTip, epoch)	3✔
492	}, func() {})	3✔
493	}
494
495	// GetLookoutTip retrieves the current lookout tip block epoch from the tower
496	// database.
497	func (t TowerDB) GetLookoutTip() (chainntnfs.BlockEpoch, error) {	3✔
498	var epoch *chainntnfs.BlockEpoch	3✔
499	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	6✔
500	lookoutTip := tx.ReadBucket(lookoutTipBkt)	3✔
501	if lookoutTip == nil {	3✔
502	return ErrUninitializedDB	×
503	}	×
504
505	epoch = getLookoutEpoch(lookoutTip)	3✔
506		3✔
507	return nil	3✔
508	}, func() {	3✔
509	epoch = nil	3✔
510	})	3✔
511	if err != nil {	3✔
512	return nil, err	×
513	}	×
514
515	return epoch, nil	3✔
516	}
517
518	// getSession retrieves the session info from the sessions bucket identified by
519	// its session id. An error is returned if the session is not found or a
520	// deserialization error occurs.
521	func getSession(sessions kvdb.RBucket, id []byte) (*SessionInfo, error) {	3✔
522	sessionBytes := sessions.Get(id)	3✔
523	if sessionBytes == nil {	6✔
524	return nil, ErrSessionNotFound	3✔
525	}	3✔
526
527	var session SessionInfo	3✔
528	err := session.Decode(bytes.NewReader(sessionBytes))	3✔
529	if err != nil {	3✔
530	return nil, err	×
531	}	×
532
533	return &session, nil	3✔
534	}
535
536	// putSession stores the session info in the sessions bucket identified by its
537	// session id. An error is returned if a serialization error occurs.
538	func putSession(sessions kvdb.RwBucket, session *SessionInfo) error {	3✔
539	var b bytes.Buffer	3✔
540	err := session.Encode(&b)	3✔
541	if err != nil {	3✔
542	return err	×
543	}	×
544
545	return sessions.Put(session.ID[:], b.Bytes())	3✔
546	}
547
548	// touchSessionHintBkt initializes the session-hint bucket for a particular
549	// session id. This ensures that future calls to getHintsForSession or
550	// putHintForSession can rely on the bucket already being created, and fail if
551	// index has not been initialized as this points to improper usage.
552	func touchSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {	3✔
553	_, err := updateIndex.CreateBucketIfNotExists(id[:])	3✔
554	return err	3✔
555	}	3✔
556
557	// removeSessionHintBkt prunes the session-hint bucket for the given session id
558	// and all of the hints contained inside. This should be used to clean up the
559	// index upon session deletion.
560	func removeSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {	3✔
561	return updateIndex.DeleteNestedBucket(id[:])	3✔
562	}	3✔
563
564	// getHintsForSession returns all known hints belonging to the given session id.
565	// If the index for the session has not been initialized, this method returns
566	// ErrNoSessionHintIndex.
567	func getHintsForSession(updateIndex kvdb.RBucket,
568	id *SessionID) ([]blob.BreachHint, error) {	3✔
569		3✔
570	sessionHints := updateIndex.NestedReadBucket(id[:])	3✔
571	if sessionHints == nil {	3✔
572	return nil, ErrNoSessionHintIndex	×
573	}	×
574
575	var hints []blob.BreachHint	3✔
576	err := sessionHints.ForEach(func(k, _ []byte) error {	6✔
577	if len(k) != blob.BreachHintSize {	3✔
578	return nil	×
579	}	×
580
581	var hint blob.BreachHint	3✔
582	copy(hint[:], k)	3✔
583	hints = append(hints, hint)	3✔
584	return nil	3✔
585	})
586	if err != nil {	3✔
587	return nil, err	×
588	}	×
589
590	return hints, nil	3✔
591	}
592
593	// putHintForSession inserts a record into the update index for a given
594	// (session, hint) pair. The hints are coalesced under a bucket for the target
595	// session id, and used to perform efficient removal of updates. If the index
596	// for the session has not been initialized, this method returns
597	// ErrNoSessionHintIndex.
598	func putHintForSession(updateIndex kvdb.RwBucket, id *SessionID,
599	hint blob.BreachHint) error {	3✔
600		3✔
601	sessionHints := updateIndex.NestedReadWriteBucket(id[:])	3✔
602	if sessionHints == nil {	3✔
603	return ErrNoSessionHintIndex	×
604	}	×
605
606	return sessionHints.Put(hint[:], []byte{})	3✔
607	}
608
609	// putLookoutEpoch stores the given lookout tip block epoch in provided bucket.
610	func putLookoutEpoch(bkt kvdb.RwBucket, epoch *chainntnfs.BlockEpoch) error {	3✔
611	epochBytes := make([]byte, 36)	3✔
612	copy(epochBytes, epoch.Hash[:])	3✔
613	byteOrder.PutUint32(epochBytes[32:], uint32(epoch.Height))	3✔
614		3✔
615	return bkt.Put(lookoutTipKey, epochBytes)	3✔
616	}	3✔
617
618	// getLookoutEpoch retrieves the lookout tip block epoch from the given bucket.
619	// A nil epoch is returned if no update exists.
620	func getLookoutEpoch(bkt kvdb.RBucket) *chainntnfs.BlockEpoch {	3✔
621	epochBytes := bkt.Get(lookoutTipKey)	3✔
622	if len(epochBytes) != 36 {	6✔
623	return nil	3✔
624	}	3✔
625
UNCOV 626	var hash chainhash.Hash	×
UNCOV 627	copy(hash[:], epochBytes[:32])	×
UNCOV 628	height := byteOrder.Uint32(epochBytes[32:])	×
UNCOV 629		×
UNCOV 630	return &chainntnfs.BlockEpoch{	×
UNCOV 631	Hash: &hash,	×
UNCOV 632	Height: int32(height),	×
UNCOV 633	}	×
634	}
635
636	// errBucketNotEmpty is a helper error returned when testing whether a bucket is
637	// empty or not.
638	var errBucketNotEmpty = errors.New("bucket not empty")
639
640	// isBucketEmpty returns errBucketNotEmpty if the bucket is not empty.
641	func isBucketEmpty(bkt kvdb.RBucket) error {	3✔
642	return bkt.ForEach(func(_, _ []byte) error {	3✔
UNCOV 643	return errBucketNotEmpty	×
UNCOV 644	})	×
645	}

lightningnetwork / lnd / 13566028875

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

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