Merge pull request #5635 from bhandras/flatten-htlc-bucket

channeldb: flatten the htlc attempts bucket
This commit is contained in:
Olaoluwa Osuntokun authored and GitHub committed 2021-09-07 20:10:12 -07:00
commit 75f5b407ea
6 files changed
+465 -85

No files matched your search

+5
View File
@@ -17,6 +17,7 @@ import (
"github.com/lightningnetwork/lnd/channeldb/migration16" "github.com/lightningnetwork/lnd/channeldb/migration16"
"github.com/lightningnetwork/lnd/channeldb/migration20" "github.com/lightningnetwork/lnd/channeldb/migration20"
"github.com/lightningnetwork/lnd/channeldb/migration21" "github.com/lightningnetwork/lnd/channeldb/migration21"
"github.com/lightningnetwork/lnd/channeldb/migration23"
"github.com/lightningnetwork/lnd/channeldb/migration_01_to_11" "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
"github.com/lightningnetwork/lnd/clock" "github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/kvdb"
@@ -193,6 +194,10 @@ var (
number: 22, number: 22,
migration: mig.CreateTLB(setIDIndexBucket), migration: mig.CreateTLB(setIDIndexBucket),
}, },
{
number: 23,
migration: migration23.MigrateHtlcAttempts,
},
} }
// Big endian is the preferred byte order, due to cursor scans over // Big endian is the preferred byte order, due to cursor scans over
+163
View File
@@ -0,0 +1,163 @@
package migration23
import (
"fmt"
"github.com/lightningnetwork/lnd/kvdb"
)
var (
// paymentsRootBucket is the name of the top-level bucket within the
// database that stores all data related to payments.
paymentsRootBucket = []byte("payments-root-bucket")
// paymentHtlcsBucket is a bucket where we'll store the information
// about the HTLCs that were attempted for a payment.
paymentHtlcsBucket = []byte("payment-htlcs-bucket")
// oldAttemptInfoKey is a key used in a HTLC's sub-bucket to store the
// info about the attempt that was done for the HTLC in question.
oldAttemptInfoKey = []byte("htlc-attempt-info")
// oldSettleInfoKey is a key used in a HTLC's sub-bucket to store the
// settle info, if any.
oldSettleInfoKey = []byte("htlc-settle-info")
// oldFailInfoKey is a key used in a HTLC's sub-bucket to store
// failure information, if any.
oldFailInfoKey = []byte("htlc-fail-info")
// htlcAttemptInfoKey is the key used as the prefix of an HTLC attempt
// to store the info about the attempt that was done for the HTLC in
// question. The HTLC attempt ID is concatenated at the end.
htlcAttemptInfoKey = []byte("ai")
// htlcSettleInfoKey is the key used as the prefix of an HTLC attempt
// settle info, if any. The HTLC attempt ID is concatenated at the end.
htlcSettleInfoKey = []byte("si")
// htlcFailInfoKey is the key used as the prefix of an HTLC attempt
// failure information, if any.The HTLC attempt ID is concatenated at
// the end.
htlcFailInfoKey = []byte("fi")
)
// htlcBucketKey creates a composite key from prefix and id where the result is
// simply the two concatenated. This is the exact copy from payments.go.
func htlcBucketKey(prefix, id []byte) []byte {
key := make([]byte, len(prefix)+len(id))
copy(key, prefix)
copy(key[len(prefix):], id)
return key
}
// MigrateHtlcAttempts will gather all htlc-attempt-info's, htlcs-settle-info's
// and htlcs-fail-info's from the attempt ID buckes and re-store them using the
// flattened keys to each payment's payment-htlcs-bucket.
func MigrateHtlcAttempts(tx kvdb.RwTx) error {
payments := tx.ReadWriteBucket(paymentsRootBucket)
if payments == nil {
return nil
}
// Collect all payment hashes so we can migrate payments one-by-one to
// avoid any bugs bbolt might have when invalidating cursors.
// For 100 million payments, this would need about 3 GiB memory so we
// should hopefully be fine for very large nodes too.
var paymentHashes []string
if err := payments.ForEach(func(hash, v []byte) error {
// Get the bucket which contains the payment, fail if the key
// does not have a bucket.
bucket := payments.NestedReadBucket(hash)
if bucket == nil {
return fmt.Errorf("key must be a bucket: '%v'",
string(paymentsRootBucket))
}
paymentHashes = append(paymentHashes, string(hash))
return nil
}); err != nil {
return err
}
for _, paymentHash := range paymentHashes {
payment := payments.NestedReadWriteBucket([]byte(paymentHash))
if payment.Get(paymentHtlcsBucket) != nil {
return fmt.Errorf("key must be a bucket: '%v'",
string(paymentHtlcsBucket))
}
htlcs := payment.NestedReadWriteBucket(paymentHtlcsBucket)
if htlcs == nil {
// Nothing to migrate for this payment.
continue
}
if err := migrateHtlcsBucket(htlcs); err != nil {
return err
}
}
return nil
}
// migrateHtlcsBucket is a helper to gather, transform and re-store htlc attempt
// key/values.
func migrateHtlcsBucket(htlcs kvdb.RwBucket) error {
// Collect attempt ids so that we can migrate attempts one-by-one
// to avoid any bugs bbolt might have when invalidating cursors.
var aids []string
// First we collect all htlc attempt ids.
if err := htlcs.ForEach(func(aid, v []byte) error {
aids = append(aids, string(aid))
return nil
}); err != nil {
return err
}
// Next we go over these attempts, fetch all data and migrate.
for _, aid := range aids {
aidKey := []byte(aid)
attempt := htlcs.NestedReadWriteBucket(aidKey)
if attempt == nil {
return fmt.Errorf("non bucket element '%v' in '%v' "+
"bucket", aidKey, string(paymentHtlcsBucket))
}
// Collect attempt/settle/fail infos.
attemptInfo := attempt.Get(oldAttemptInfoKey)
if len(attemptInfo) > 0 {
newKey := htlcBucketKey(htlcAttemptInfoKey, aidKey)
if err := htlcs.Put(newKey, attemptInfo); err != nil {
return err
}
}
settleInfo := attempt.Get(oldSettleInfoKey)
if len(settleInfo) > 0 {
newKey := htlcBucketKey(htlcSettleInfoKey, aidKey)
if err := htlcs.Put(newKey, settleInfo); err != nil {
return err
}
}
failInfo := attempt.Get(oldFailInfoKey)
if len(failInfo) > 0 {
newKey := htlcBucketKey(htlcFailInfoKey, aidKey)
if err := htlcs.Put(newKey, failInfo); err != nil {
return err
}
}
}
// Finally we delete old attempt buckets.
for _, aid := range aids {
if err := htlcs.DeleteNestedBucket([]byte(aid)); err != nil {
return err
}
}
return nil
}
+176
View File
@@ -0,0 +1,176 @@
package migration23
import (
"testing"
"github.com/lightningnetwork/lnd/channeldb/migtest"
"github.com/lightningnetwork/lnd/kvdb"
)
var (
hexStr = migtest.Hex
hash1Str = "02acee76ebd53d00824410cf6adecad4f50334dac702bd5a2d3ba01b91709f0e"
hash1 = hexStr(hash1Str)
paymentID1 = hexStr("0000000000000001")
attemptID1 = hexStr("0000000000000001")
attemptID2 = hexStr("0000000000000002")
hash2Str = "62eb3f0a48f954e495d0c14ac63df04a67cefa59dafdbcd3d5046d1f5647840c"
hash2 = hexStr(hash2Str)
paymentID2 = hexStr("0000000000000002")
attemptID3 = hexStr("0000000000000003")
hash3Str = "99eb3f0a48f954e495d0c14ac63df04af8cefa59dafdbcd3d5046d1f564784d1"
hash3 = hexStr(hash3Str)
// failing1 will fail because all payment hashes should point to sub
// buckets containing payment data.
failing1 = map[string]interface{}{
hash1: "bogus",
}
// failing2 will fail because the "payment-htlcs-bucket" key must point
// to an actual bucket or be non-existent, but never point to a value.
failing2 = map[string]interface{}{
hash1: map[string]interface{}{
"payment-htlcs-bucket": "bogus",
},
}
// failing3 will fail because each attempt ID inside the
// "payment-htlcs-bucket" must point to a sub-bucket.
failing3 = map[string]interface{}{
hash1: map[string]interface{}{
"payment-creation-info": "aaaa",
"payment-fail-info": "bbbb",
"payment-htlcs-bucket": map[string]interface{}{
attemptID1: map[string]interface{}{
"htlc-attempt-info": "cccc",
"htlc-fail-info": "dddd",
},
attemptID2: "bogus",
},
"payment-sequence-key": paymentID1,
},
}
// pre is a sample snapshot (with fake values) before migration.
pre = map[string]interface{}{
hash1: map[string]interface{}{
"payment-creation-info": "aaaa",
"payment-fail-info": "bbbb",
"payment-htlcs-bucket": map[string]interface{}{
attemptID1: map[string]interface{}{
"htlc-attempt-info": "cccc",
"htlc-fail-info": "dddd",
},
},
"payment-sequence-key": paymentID1,
},
hash2: map[string]interface{}{
"payment-creation-info": "eeee",
"payment-htlcs-bucket": map[string]interface{}{
attemptID2: map[string]interface{}{
"htlc-attempt-info": "ffff",
"htlc-fail-info": "gggg",
},
attemptID3: map[string]interface{}{
"htlc-attempt-info": "hhhh",
"htlc-settle-info": "iiii",
},
},
"payment-sequence-key": paymentID2,
},
hash3: map[string]interface{}{
"payment-creation-info": "aaaa",
"payment-fail-info": "bbbb",
"payment-sequence-key": paymentID1,
},
}
// post is the expected data after migration.
post = map[string]interface{}{
hash1: map[string]interface{}{
"payment-creation-info": "aaaa",
"payment-fail-info": "bbbb",
"payment-htlcs-bucket": map[string]interface{}{
"ai" + attemptID1: "cccc",
"fi" + attemptID1: "dddd",
},
"payment-sequence-key": paymentID1,
},
hash2: map[string]interface{}{
"payment-creation-info": "eeee",
"payment-htlcs-bucket": map[string]interface{}{
"ai" + attemptID2: "ffff",
"fi" + attemptID2: "gggg",
"ai" + attemptID3: "hhhh",
"si" + attemptID3: "iiii",
},
"payment-sequence-key": paymentID2,
},
hash3: map[string]interface{}{
"payment-creation-info": "aaaa",
"payment-fail-info": "bbbb",
"payment-sequence-key": paymentID1,
},
}
)
// TestMigrateHtlcAttempts tests that migration htlc attempts to the flattened
// structure succeeds.
func TestMigrateHtlcAttempts(t *testing.T) {
var paymentsRootBucket = []byte("payments-root-bucket")
tests := []struct {
name string
shouldFail bool
pre map[string]interface{}
post map[string]interface{}
}{
{
name: "migration ok",
shouldFail: false,
pre: pre,
post: post,
},
{
name: "non-bucket payments-root-bucket",
shouldFail: true,
pre: failing1,
post: failing1,
},
{
name: "non-bucket payment-htlcs-bucket",
shouldFail: true,
pre: failing2,
post: failing2,
},
{
name: "non-bucket htlc attempt",
shouldFail: true,
pre: failing3,
post: failing3,
},
}
for _, test := range tests {
test := test
migtest.ApplyMigration(
t,
func(tx kvdb.RwTx) error {
return migtest.RestoreDB(
tx, paymentsRootBucket, test.pre,
)
},
func(tx kvdb.RwTx) error {
return migtest.VerifyDB(
tx, paymentsRootBucket, test.post,
)
},
MigrateHtlcAttempts,
test.shouldFail,
)
}
}
+10 -15
View File
@@ -368,14 +368,10 @@ func (p *PaymentControl) RegisterAttempt(paymentHash lntypes.Hash,
return err return err
} }
// Create bucket for this attempt. Fail if the bucket already err = htlcsBucket.Put(
// exists. htlcBucketKey(htlcAttemptInfoKey, htlcIDBytes),
htlcBucket, err := htlcsBucket.CreateBucket(htlcIDBytes) htlcInfoBytes,
if err != nil { )
return err
}
err = htlcBucket.Put(htlcAttemptInfoKey, htlcInfoBytes)
if err != nil { if err != nil {
return err return err
} }
@@ -427,8 +423,8 @@ func (p *PaymentControl) FailAttempt(hash lntypes.Hash,
func (p *PaymentControl) updateHtlcKey(paymentHash lntypes.Hash, func (p *PaymentControl) updateHtlcKey(paymentHash lntypes.Hash,
attemptID uint64, key, value []byte) (*MPPayment, error) { attemptID uint64, key, value []byte) (*MPPayment, error) {
htlcIDBytes := make([]byte, 8) aid := make([]byte, 8)
binary.BigEndian.PutUint64(htlcIDBytes, attemptID) binary.BigEndian.PutUint64(aid, attemptID)
var payment *MPPayment var payment *MPPayment
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
@@ -456,23 +452,22 @@ func (p *PaymentControl) updateHtlcKey(paymentHash lntypes.Hash,
return fmt.Errorf("htlcs bucket not found") return fmt.Errorf("htlcs bucket not found")
} }
htlcBucket := htlcsBucket.NestedReadWriteBucket(htlcIDBytes) if htlcsBucket.Get(htlcBucketKey(htlcAttemptInfoKey, aid)) == nil {
if htlcBucket == nil {
return fmt.Errorf("HTLC with ID %v not registered", return fmt.Errorf("HTLC with ID %v not registered",
attemptID) attemptID)
} }
// Make sure the shard is not already failed or settled. // Make sure the shard is not already failed or settled.
if htlcBucket.Get(htlcFailInfoKey) != nil { if htlcsBucket.Get(htlcBucketKey(htlcFailInfoKey, aid)) != nil {
return ErrAttemptAlreadyFailed return ErrAttemptAlreadyFailed
} }
if htlcBucket.Get(htlcSettleInfoKey) != nil { if htlcsBucket.Get(htlcBucketKey(htlcSettleInfoKey, aid)) != nil {
return ErrAttemptAlreadySettled return ErrAttemptAlreadySettled
} }
// Add or update the key for this htlc. // Add or update the key for this htlc.
err = htlcBucket.Put(key, value) err = htlcsBucket.Put(htlcBucketKey(key, aid), value)
if err != nil { if err != nil {
return err return err
} }
+104 -70
View File
@@ -35,14 +35,11 @@ var (
// | | // | |
// | |--payment-htlcs-bucket (shard-bucket) // | |--payment-htlcs-bucket (shard-bucket)
// | | | // | | |
// | | |-- <htlc attempt ID> // | | |-- ai<htlc attempt ID>: <htlc attempt info>
// | | | |--htlc-attempt-info-key: <htlc attempt info> // | | |-- si<htlc attempt ID>: <(optional) settle info>
// | | | |--htlc-settle-info-key: <(optional) settle info> // | | |-- fi<htlc attempt ID>: <(optional) fail info>
// | | | |--htlc-fail-info-key: <(optional) fail info>
// | | | // | | |
// | | |-- <htlc attempt ID> // | | ...
// | | | |
// | | ... ...
// | | // | |
// | | // | |
// | |--duplicate-bucket (only for old, completed payments) // | |--duplicate-bucket (only for old, completed payments)
@@ -50,9 +47,9 @@ var (
// | |-- <seq-num> // | |-- <seq-num>
// | | |--sequence-key: <sequence number> // | | |--sequence-key: <sequence number>
// | | |--creation-info-key: <creation info> // | | |--creation-info-key: <creation info>
// | | |--attempt-info-key: <attempt info> // | | |--ai: <attempt info>
// | | |--settle-info-key: <settle info> // | | |--si: <settle info>
// | | |--fail-info-key: <fail info> // | | |--fi: <fail info>
// | | // | |
// | |-- <seq-num> // | |-- <seq-num>
// | | | // | | |
@@ -77,17 +74,19 @@ var (
// about the HTLCs that were attempted for a payment. // about the HTLCs that were attempted for a payment.
paymentHtlcsBucket = []byte("payment-htlcs-bucket") paymentHtlcsBucket = []byte("payment-htlcs-bucket")
// htlcAttemptInfoKey is a key used in a HTLC's sub-bucket to store the // htlcAttemptInfoKey is the key used as the prefix of an HTLC attempt
// info about the attempt that was done for the HTLC in question. // to store the info about the attempt that was done for the HTLC in
htlcAttemptInfoKey = []byte("htlc-attempt-info") // question. The HTLC attempt ID is concatenated at the end.
htlcAttemptInfoKey = []byte("ai")
// htlcSettleInfoKey is a key used in a HTLC's sub-bucket to store the // htlcSettleInfoKey is the key used as the prefix of an HTLC attempt
// settle info, if any. // settle info, if any. The HTLC attempt ID is concatenated at the end.
htlcSettleInfoKey = []byte("htlc-settle-info") htlcSettleInfoKey = []byte("si")
// htlcFailInfoKey is a key used in a HTLC's sub-bucket to store // htlcFailInfoKey is the key used as the prefix of an HTLC attempt
// failure information, if any. // failure information, if any.The HTLC attempt ID is concatenated at
htlcFailInfoKey = []byte("htlc-fail-info") // the end.
htlcFailInfoKey = []byte("fi")
// paymentFailInfoKey is a key used in the payment's sub-bucket to // paymentFailInfoKey is a key used in the payment's sub-bucket to
// store information about the reason a payment failed. // store information about the reason a payment failed.
@@ -230,6 +229,15 @@ type PaymentCreationInfo struct {
PaymentRequest []byte PaymentRequest []byte
} }
// htlcBucketKey creates a composite key from prefix and id where the result is
// simply the two concatenated.
func htlcBucketKey(prefix, id []byte) []byte {
key := make([]byte, len(prefix)+len(id))
copy(key, prefix)
copy(key[len(prefix):], id)
return key
}
// FetchPayments returns all sent payments found in the DB. // FetchPayments returns all sent payments found in the DB.
// //
// nolint: dupl // nolint: dupl
@@ -378,80 +386,93 @@ func fetchPayment(bucket kvdb.RBucket) (*MPPayment, error) {
// fetchHtlcAttempts retrives all htlc attempts made for the payment found in // fetchHtlcAttempts retrives all htlc attempts made for the payment found in
// the given bucket. // the given bucket.
func fetchHtlcAttempts(bucket kvdb.RBucket) ([]HTLCAttempt, error) { func fetchHtlcAttempts(bucket kvdb.RBucket) ([]HTLCAttempt, error) {
htlcs := make([]HTLCAttempt, 0) htlcsMap := make(map[uint64]*HTLCAttempt)
err := bucket.ForEach(func(k, _ []byte) error { attemptInfoCount := 0
aid := byteOrder.Uint64(k) err := bucket.ForEach(func(k, v []byte) error {
htlcBucket := bucket.NestedReadBucket(k) aid := byteOrder.Uint64(k[len(k)-8:])
attemptInfo, err := fetchHtlcAttemptInfo( if _, ok := htlcsMap[aid]; !ok {
htlcBucket, htlcsMap[aid] = &HTLCAttempt{}
)
if err != nil {
return err
}
attemptInfo.AttemptID = aid
htlc := HTLCAttempt{
HTLCAttemptInfo: *attemptInfo,
} }
// Settle info might be nil. var err error
htlc.Settle, err = fetchHtlcSettleInfo(htlcBucket) switch {
if err != nil { case bytes.HasPrefix(k, htlcAttemptInfoKey):
return err attemptInfo, err := readHtlcAttemptInfo(v)
if err != nil {
return err
}
attemptInfo.AttemptID = aid
htlcsMap[aid].HTLCAttemptInfo = *attemptInfo
attemptInfoCount++
case bytes.HasPrefix(k, htlcSettleInfoKey):
htlcsMap[aid].Settle, err = readHtlcSettleInfo(v)
if err != nil {
return err
}
case bytes.HasPrefix(k, htlcFailInfoKey):
htlcsMap[aid].Failure, err = readHtlcFailInfo(v)
if err != nil {
return err
}
default:
return fmt.Errorf("unknown htlc attempt key")
} }
// Failure info might be nil.
htlc.Failure, err = fetchHtlcFailInfo(htlcBucket)
if err != nil {
return err
}
htlcs = append(htlcs, htlc)
return nil return nil
}) })
if err != nil { if err != nil {
return nil, err return nil, err
} }
return htlcs, nil // Sanity check that all htlcs have an attempt info.
} if attemptInfoCount != len(htlcsMap) {
// fetchHtlcAttemptInfo fetches the payment attempt info for this htlc from the
// bucket.
func fetchHtlcAttemptInfo(bucket kvdb.RBucket) (*HTLCAttemptInfo, error) {
b := bucket.Get(htlcAttemptInfoKey)
if b == nil {
return nil, errNoAttemptInfo return nil, errNoAttemptInfo
} }
keys := make([]uint64, len(htlcsMap))
i := 0
for k := range htlcsMap {
keys[i] = k
i++
}
// Sort HTLC attempts by their attempt ID. This is needed because in the
// DB we store the attempts with keys prefixed by their status which
// changes order (groups them together by status).
sort.Slice(keys, func(i, j int) bool {
return keys[i] < keys[j]
})
htlcs := make([]HTLCAttempt, len(htlcsMap))
for i, key := range keys {
htlcs[i] = *htlcsMap[key]
}
return htlcs, nil
}
// readHtlcAttemptInfo reads the payment attempt info for this htlc.
func readHtlcAttemptInfo(b []byte) (*HTLCAttemptInfo, error) {
r := bytes.NewReader(b) r := bytes.NewReader(b)
return deserializeHTLCAttemptInfo(r) return deserializeHTLCAttemptInfo(r)
} }
// fetchHtlcSettleInfo retrieves the settle info for the htlc. If the htlc isn't // readHtlcSettleInfo reads the settle info for the htlc. If the htlc isn't
// settled, nil is returned. // settled, nil is returned.
func fetchHtlcSettleInfo(bucket kvdb.RBucket) (*HTLCSettleInfo, error) { func readHtlcSettleInfo(b []byte) (*HTLCSettleInfo, error) {
b := bucket.Get(htlcSettleInfoKey)
if b == nil {
// Settle info is optional.
return nil, nil
}
r := bytes.NewReader(b) r := bytes.NewReader(b)
return deserializeHTLCSettleInfo(r) return deserializeHTLCSettleInfo(r)
} }
// fetchHtlcFailInfo retrieves the failure info for the htlc. If the htlc hasn't // readHtlcFailInfo reads the failure info for the htlc. If the htlc hasn't
// failed, nil is returned. // failed, nil is returned.
func fetchHtlcFailInfo(bucket kvdb.RBucket) (*HTLCFailInfo, error) { func readHtlcFailInfo(b []byte) (*HTLCFailInfo, error) {
b := bucket.Get(htlcFailInfoKey)
if b == nil {
// Fail info is optional.
return nil, nil
}
r := bytes.NewReader(b) r := bytes.NewReader(b)
return deserializeHTLCFailInfo(r) return deserializeHTLCFailInfo(r)
} }
@@ -797,8 +818,21 @@ func (d *DB) DeletePayments(failedOnly, failedHtlcsOnly bool) error {
) )
for _, aid := range htlcIDs { for _, aid := range htlcIDs {
err := htlcsBucket.DeleteNestedBucket(aid) if err := htlcsBucket.Delete(
if err != nil { htlcBucketKey(htlcAttemptInfoKey, aid),
); err != nil {
return err
}
if err := htlcsBucket.Delete(
htlcBucketKey(htlcFailInfoKey, aid),
); err != nil {
return err
}
if err := htlcsBucket.Delete(
htlcBucketKey(htlcSettleInfoKey, aid),
); err != nil {
return err return err
} }
} }
@@ -211,6 +211,13 @@ you.
to make it less likely that we retry etcd transactions and make the commit to make it less likely that we retry etcd transactions and make the commit
queue more scalable. queue more scalable.
* [Flatten the payment-htlcs-bucket](https://github.com/lightningnetwork/lnd/pull/5635)
in order to make it possible to prefetch all htlc attempts of a payment in one
DB operation. Migration may fail for extremely large DBs with many payments
(10+ million). Be careful and backup your `channel.db` if you have that many
payments. Deleting all failed payments beforehand makes migration safer and
faster too.
## Performance improvements ## Performance improvements
* [Update MC store in blocks](https://github.com/lightningnetwork/lnd/pull/5515) * [Update MC store in blocks](https://github.com/lightningnetwork/lnd/pull/5515)