From 2c07da6e8e680fca783ffb7296df51ca6fe8a783 Mon Sep 17 00:00:00 2001 From: Ryan Staudt Date: Tue, 25 May 2021 13:11:28 -0500 Subject: [PATCH] multi: Update UtxoBackend to use leveldb directly. This updates the UtxoBackend implementation to use leveldb directly rather than using the database package. This results in reduced memory usage and improved processing time since it avoids the overhead of things that the database package provides, such as caching with periodic flushing, that are not needed for the UtxoBackend since it has its own caching layer (UtxoCache). An overview of the changes is as follows: - Introduce simple UTXO key sets to be used for constructing prefixes of database keys rather than using the "buckets" concept that the database package uses - All keys in the UTXO backend now start with a serialized prefix consisting of the key set (1 byte) and version of that key set (1 byte) - Add migration to move all keys in the UTXO backend from buckets to the newly defined key sets - Add Get function to the UtxoBackend interface and corresponding implementation in LevelDbUtxoBackend to allow for directly retrieving the value for a given key - Add NewIterator function to the UtxoBackend interface and corresponding implementation in LevelDbUtxoBackend to allow for iterating over the key/value pairs in the UtxoBackend - Add Update function to the UtxoBackend interface and corresponding implementation in LevelDbUtxoBackend to allow for directly updating values in the UtxoBackend in the context of a transaction - Update LevelDbUtxoBackend to use an underlying leveldb.DB instance rather than a database.DB instance - Update all LevelDbUtxoBackend methods to use the Get, NewIterator, and Update functions rather than the database package functions - Update all database queries to use keys with the new key set prefixes rather than bucketized keys - Update dbFetchUtxoBackendInfo to check the legacy bucket for database info if it doesn't exist in the new location - Update upgrade logic to use the UtxoBackend interface rather than the concrete LevelDbUtxoBackend type - Update all tests to create leveldb.DB databases instead of database.DB databases for the UtxoBackend --- blockchain/common_test.go | 35 +- blockchain/example_test.go | 21 +- blockchain/fullblocks_test.go | 35 +- blockchain/upgrade.go | 447 ++++++++++++++++++++++-- blockchain/utxobackend.go | 600 ++++++++++++++++++++------------- blockchain/utxobackend_test.go | 41 +-- blockchain/utxoio.go | 27 +- blockchain/validate_test.go | 3 +- internal/netsync/manager.go | 4 +- server.go | 3 +- 10 files changed, 898 insertions(+), 318 deletions(-) diff --git a/blockchain/common_test.go b/blockchain/common_test.go index 47238716..1f56b6cf 100644 --- a/blockchain/common_test.go +++ b/blockchain/common_test.go @@ -15,6 +15,7 @@ import ( "math" mrand "math/rand" "os" + "path/filepath" "reflect" "testing" "time" @@ -30,6 +31,9 @@ import ( "github.com/decred/dcrd/txscript/v4" "github.com/decred/dcrd/txscript/v4/sign" "github.com/decred/dcrd/wire" + "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/filter" + "github.com/syndtr/goleveldb/leveldb/opt" ) const ( @@ -100,6 +104,34 @@ func createTestDatabase(dbName string, dbType string, net wire.CurrencyNet) (dat return db, teardown, nil } +// createTestUtxoDatabase creates a test UTXO database with the provided +// database name. +func createTestUtxoDatabase(dbName string) (*leveldb.DB, func(), error) { + // Construct the database filepath and remove all from that path. + dbPath := filepath.Join(os.TempDir(), dbName) + _ = os.RemoveAll(dbPath) + + // Open the database (will create it if needed). + opts := opt.Options{ + Strict: opt.DefaultStrict, + Compression: opt.NoCompression, + Filter: filter.NewBloomFilter(10), + } + db, err := leveldb.OpenFile(dbPath, &opts) + if err != nil { + return nil, nil, err + } + + // Setup a teardown function for cleaning up. This function is returned to + // the caller to be invoked when it is done testing. + teardown := func() { + _ = db.Close() + _ = os.RemoveAll(dbPath) + } + + return db, teardown, nil +} + // chainSetup is used to create a new db and chain instance with the genesis // block already inserted. In addition to the new chain instance, it returns // a teardown function the caller should invoke when done testing to clean up. @@ -115,8 +147,7 @@ func chainSetup(dbName string, params *chaincfg.Params) (*BlockChain, func(), er } // Create a test UTXO database. - utxoDb, teardownUtxoDb, err := createTestDatabase(dbName+"_utxo", testDbType, - blockDataNet) + utxoDb, teardownUtxoDb, err := createTestUtxoDatabase(dbName + "_utxo") if err != nil { teardownDb() return nil, nil, err diff --git a/blockchain/example_test.go b/blockchain/example_test.go index f3c0d5ba..92921228 100644 --- a/blockchain/example_test.go +++ b/blockchain/example_test.go @@ -16,6 +16,9 @@ import ( "github.com/decred/dcrd/database/v2" _ "github.com/decred/dcrd/database/v2/ffldb" "github.com/decred/dcrd/dcrutil/v4" + "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/filter" + "github.com/syndtr/goleveldb/leveldb/opt" ) // This example demonstrates how to create a new chain instance and use @@ -40,6 +43,22 @@ func ExampleBlockChain_ProcessBlock() { defer os.RemoveAll(dbPath) defer db.Close() + // Additionally, create a new database to store the UTXO set. + utxoDbPath := filepath.Join(os.TempDir(), "exampleprocessblockutxodb") + _ = os.RemoveAll(utxoDbPath) + opts := opt.Options{ + Strict: opt.DefaultStrict, + Compression: opt.NoCompression, + Filter: filter.NewBloomFilter(10), + } + utxoDb, err := leveldb.OpenFile(utxoDbPath, &opts) + if err != nil { + fmt.Printf("Failed to create UTXO database: %v\n", err) + return + } + defer os.RemoveAll(utxoDbPath) + defer utxoDb.Close() + // Create a new BlockChain instance using the underlying database for // the main bitcoin network. This example does not demonstrate some // of the other available configuration options such as specifying a @@ -53,7 +72,7 @@ func ExampleBlockChain_ProcessBlock() { ChainParams: mainNetParams, TimeSource: blockchain.NewMedianTime(), UtxoCache: blockchain.NewUtxoCache(&blockchain.UtxoCacheConfig{ - Backend: blockchain.NewLevelDbUtxoBackend(db), + Backend: blockchain.NewLevelDbUtxoBackend(utxoDb), FlushBlockDB: func() error { return nil }, diff --git a/blockchain/fullblocks_test.go b/blockchain/fullblocks_test.go index dbb48496..c85d1fbc 100644 --- a/blockchain/fullblocks_test.go +++ b/blockchain/fullblocks_test.go @@ -12,6 +12,7 @@ import ( "fmt" "io/ioutil" "os" + "path/filepath" "testing" "github.com/decred/dcrd/blockchain/v4" @@ -22,6 +23,9 @@ import ( "github.com/decred/dcrd/dcrutil/v4" "github.com/decred/dcrd/txscript/v4" "github.com/decred/dcrd/wire" + "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/filter" + "github.com/syndtr/goleveldb/leveldb/opt" ) const ( @@ -92,6 +96,34 @@ func createTestDatabase(dbName string, dbType string, net wire.CurrencyNet) (dat return db, teardown, nil } +// createTestUtxoDatabase creates a test UTXO database with the provided +// database name. +func createTestUtxoDatabase(dbName string) (*leveldb.DB, func(), error) { + // Construct the database filepath and remove all from that path. + dbPath := filepath.Join(os.TempDir(), dbName) + _ = os.RemoveAll(dbPath) + + // Open the database (will create it if needed). + opts := opt.Options{ + Strict: opt.DefaultStrict, + Compression: opt.NoCompression, + Filter: filter.NewBloomFilter(10), + } + db, err := leveldb.OpenFile(dbPath, &opts) + if err != nil { + return nil, nil, err + } + + // Setup a teardown function for cleaning up. This function is returned to + // the caller to be invoked when it is done testing. + teardown := func() { + _ = db.Close() + _ = os.RemoveAll(dbPath) + } + + return db, teardown, nil +} + // chainSetup is used to create a new db and chain instance with the genesis // block already inserted. In addition to the new chain instance, it returns // a teardown function the caller should invoke when done testing to clean up. @@ -107,8 +139,7 @@ func chainSetup(dbName string, params *chaincfg.Params) (*blockchain.BlockChain, } // Create a test UTXO database. - utxoDb, teardownUtxoDb, err := createTestDatabase(dbName+"_utxo", testDbType, - blockDataNet) + utxoDb, teardownUtxoDb, err := createTestUtxoDatabase(dbName + "_utxo") if err != nil { teardownDb() return nil, nil, err diff --git a/blockchain/upgrade.go b/blockchain/upgrade.go index 75b4dfcd..58482c90 100644 --- a/blockchain/upgrade.go +++ b/blockchain/upgrade.go @@ -384,6 +384,51 @@ func batchedUpdate(ctx context.Context, db database.DB, doBatch batchFn) error { return nil } +// utxoBackendBatchFn represents the batch function used by the UTXO backend +// batched update function. +type utxoBackendBatchFn func(tx UtxoBackendTx) (bool, error) + +// utxoBackendBatchedUpdate calls the provided batch function repeatedly until +// it either returns an error other than the special ones described in this +// comment or its return indicates no more calls are necessary. +// +// In order to ensure the backend is updated with the results of the batch that +// have already been successfully completed, it is allowed to return +// errBatchFinished and errInterruptRequested. In the case of the former, the +// error will be ignored. In the case of the latter, the backend will be +// updated and the error will be returned accordingly. The backend will NOT +// be updated if any other errors are returned. +func utxoBackendBatchedUpdate(ctx context.Context, + utxoBackend UtxoBackend, doBatch utxoBackendBatchFn) error { + + var isFullyDone bool + for !isFullyDone { + err := utxoBackend.Update(func(tx UtxoBackendTx) error { + var err error + isFullyDone, err = doBatch(tx) + if errors.Is(err, errInterruptRequested) || + errors.Is(err, errBatchFinished) { + + // No error here so the database transaction is not cancelled + // and therefore outstanding work is written to disk. The outer + // function will exit with an interrupted error below due to + // another interrupted check. + return nil + } + return err + }) + if err != nil { + return err + } + + if interruptRequested(ctx) { + return errInterruptRequested + } + } + + return nil +} + // clearFailedBlockFlagsV2 unmarks all blocks in a version 2 block index // previously marked failed so they are eligible for validation again under new // consensus rules. This ensures clients that did not update prior to new rules @@ -3191,10 +3236,38 @@ func upgradeToVersion10(ctx context.Context, db database.DB, dbInfo *databaseInf // separateUtxoDatabase moves the UTXO set and state from the block database to // the UTXO database. -func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.DB) error { - // Hardcoded bucket and key names so updates do not affect old upgrades. - v1UtxoSetStateKeyName := []byte("utxosetstate") +func separateUtxoDatabase(ctx context.Context, db database.DB, + utxoBackend UtxoBackend) error { + + // Key names, versions, and prefixes are hardcoded below so that updates do + // not affect old upgrades. + + // Legacy buckets and key names. v3UtxoSetBucketName := []byte("utxosetv3") + utxoSetStateKeyNameV1 := []byte("utxosetstate") + + // --------------------------------------------------------------------------- + // The new keys in the UTXO backend start with a serialized prefix consisting + // of the key set and version of that key set as follows: + // + // + // + // Key Value Size Description + // key set uint8 1 byte The key set identifier, as defined below + // version uint8 1 byte The version of the key set + // + // The key sets as of this migration are: + // utxoKeySetDbInfo: 1 + // utxoKeySetUtxoState: 2 + // utxoKeySetUtxoSet: 3 + // + // The versions as of this migration are: + // utxoKeySetDbInfo: 0 + // utxoKeySetUtxoState: 1 + // utxoKeySetUtxoSet: 3 + // --------------------------------------------------------------------------- + utxoSetStateKeyNew := []byte("\x02\x01utxosetstate") + utxoPrefixUtxoSetV3 := []byte("\x03\x03") log.Info("Migrating UTXO database. This may take a while...") start := time.Now() @@ -3209,7 +3282,7 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D var resumeOffset uint32 var totalMigrated uint64 var err error - doBatch := func(dbTx database.Tx, utxoDbTx database.Tx) (bool, error) { + doBatch := func(dbTx database.Tx, tx UtxoBackendTx) (bool, error) { // Get the UTXO set bucket for both the old and the new database. v3UtxoSetBucketOldDb := dbTx.Metadata().Bucket(v3UtxoSetBucketName) if v3UtxoSetBucketOldDb == nil { @@ -3217,11 +3290,6 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D // as there is nothing to do. return true, nil } - v3UtxoSetBucketNewDb := utxoDbTx.Metadata().Bucket(v3UtxoSetBucketName) - if v3UtxoSetBucketNewDb == nil { - return false, fmt.Errorf("bucket %s does not exist", - v3UtxoSetBucketName) - } // Migrate UTXO set entries so long as the max number of entries for this // batch has not been exceeded. @@ -3252,7 +3320,8 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D resumeOffset++ // Create the new entry in the V3 bucket. - err = v3UtxoSetBucketNewDb.Put(cursor.Key(), cursor.Value()) + newKey := prefixedKey(utxoPrefixUtxoSetV3, cursor.Key()) + err = tx.Put(newKey, cursor.Value()) if err != nil { return false, err } @@ -3272,9 +3341,9 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D var isFullyDone bool for !isFullyDone { err := db.View(func(dbTx database.Tx) error { - err := utxoDb.Update(func(utxoDbTx database.Tx) error { + return utxoBackend.Update(func(tx UtxoBackendTx) error { var err error - isFullyDone, err = doBatch(dbTx, utxoDbTx) + isFullyDone, err = doBatch(dbTx, tx) if errors.Is(err, errInterruptRequested) || errors.Is(err, errBatchFinished) { @@ -3282,11 +3351,10 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D // and therefore outstanding work is written to disk. The outer // function will exit with an interrupted error below due to // another interrupted check. - err = nil + return nil } return err }) - return err }) if err != nil { return err @@ -3308,7 +3376,7 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D // Get the UTXO set state from the old database. var serialized []byte err = db.View(func(dbTx database.Tx) error { - serialized = dbTx.Metadata().Get(v1UtxoSetStateKeyName) + serialized = dbTx.Metadata().Get(utxoSetStateKeyNameV1) return nil }) if err != nil { @@ -3317,25 +3385,18 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D if serialized != nil { // Set the UTXO set state in the new database. - err = utxoDb.Update(func(utxoDbTx database.Tx) error { - return utxoDbTx.Metadata().Put(v1UtxoSetStateKeyName, serialized) + err = utxoBackend.Update(func(tx UtxoBackendTx) error { + return tx.Put(utxoSetStateKeyNew, serialized) }) if err != nil { return err } } - // Force the UTXO database to flush to disk before removing the UTXO set - // and UTXO state from the block database. - err = utxoDb.Flush() - if err != nil { - return err - } - if serialized != nil { // Delete the UTXO set state from the old database. err = db.Update(func(dbTx database.Tx) error { - return dbTx.Metadata().Delete(v1UtxoSetStateKeyName) + return dbTx.Metadata().Delete(utxoSetStateKeyNameV1) }) if err != nil { return err @@ -3363,6 +3424,331 @@ func separateUtxoDatabase(ctx context.Context, db database.DB, utxoDb database.D return db.Flush() } +// fetchLegacyBucketId returns the legacy bucket id for the provided bucket +// name. A Get function must be provided to retrieve key/value pairs from the +// underlying data store. +func fetchLegacyBucketId(getFn func(key []byte) ([]byte, error), + bucketName []byte) ([]byte, error) { + + // bucketIndexPrefix is the prefix used for all entries in the bucket index. + bucketIndexPrefix := []byte("bidx") + + // parentBucketId is the parent bucket id. This is hardcoded to zero here + // since none of the legacy UTXO database buckets had parent buckets. + parentBucketId := []byte{0x00, 0x00, 0x00, 0x00} + + // Construct the key and fetch the corresponding bucket id from the database. + // The serialized bucket index key format is: + // + bucketIdKey := prefixedKey(bucketIndexPrefix, parentBucketId) + bucketIdKey = append(bucketIdKey, bucketName...) + bucketId, err := getFn(bucketIdKey) + if err != nil { + str := fmt.Sprintf("error fetching legacy bucket id for %v bucket", + string(bucketName)) + return nil, convertLdbErr(err, str) + } + + return bucketId, nil +} + +// migrateUtxoDbBuckets migrates the UTXO data to use simple key prefixes rather +// than buckets. +func migrateUtxoDbBuckets(ctx context.Context, utxoBackend UtxoBackend) error { + // Key names, versions, and prefixes are hardcoded below so that updates do + // not affect old upgrades. + + // Legacy buckets and key names. + metadataLegacyBucketID := []byte{0x00, 0x00, 0x00, 0x00} + utxoSetLegacyBucketName := []byte("utxosetv3") + utxoDbInfoLegacyBucketName := []byte("dbinfo") + utxoDbInfoVersionKeyNameV1 := []byte("version") + utxoDbInfoCompVerKeyNameV1 := []byte("compver") + utxoDbInfoUtxoVerKeyNameV1 := []byte("utxover") + utxoDbInfoCreatedKeyNameV1 := []byte("created") + utxoSetStateKeyNameV1 := []byte("utxosetstate") + + // --------------------------------------------------------------------------- + // The new keys in the UTXO backend start with a serialized prefix consisting + // of the key set and version of that key set as follows: + // + // + // + // Key Value Size Description + // key set uint8 1 byte The key set identifier, as defined below + // version uint8 1 byte The version of the key set + // + // The key sets as of this migration are: + // utxoKeySetDbInfo: 1 + // utxoKeySetUtxoState: 2 + // utxoKeySetUtxoSet: 3 + // + // The versions as of this migration are: + // utxoKeySetDbInfo: 0 + // utxoKeySetUtxoState: 1 + // utxoKeySetUtxoSet: 3 + // --------------------------------------------------------------------------- + utxoDbInfoVersionKeyNew := []byte("\x01\x00version") + utxoDbInfoCompVerKeyNew := []byte("\x01\x00compver") + utxoDbInfoUtxoVerKeyNew := []byte("\x01\x00utxover") + utxoDbInfoCreatedKeyNew := []byte("\x01\x00created") + utxoSetStateKeyNew := []byte("\x02\x01utxosetstate") + utxoPrefixUtxoSetV3 := []byte("\x03\x03") + + // moveKey is a helper function that uses an existing UTXO backend transaction + // to move a key. + moveKey := func(tx UtxoBackendTx, oldKey, newKey []byte) error { + serialized, err := tx.Get(oldKey) + if err != nil { + return err + } + err = tx.Put(newKey, serialized) + if err != nil { + return err + } + return tx.Delete(oldKey) + } + + // Move the database info from the legacy bucket to the new keys. + bucketId, err := fetchLegacyBucketId(utxoBackend.Get, + utxoDbInfoLegacyBucketName) + if err != nil { + return err + } + if bucketId != nil { + err := utxoBackend.Update(func(tx UtxoBackendTx) error { + // Move the database version from the legacy bucket to the new key. + oldKey := prefixedKey(bucketId, utxoDbInfoVersionKeyNameV1) + err = moveKey(tx, oldKey, utxoDbInfoVersionKeyNew) + if err != nil { + return fmt.Errorf("error migrating database version: %w", err) + } + + // Move the database compression version from the legacy bucket to the new + // key. + oldKey = prefixedKey(bucketId, utxoDbInfoCompVerKeyNameV1) + err = moveKey(tx, oldKey, utxoDbInfoCompVerKeyNew) + if err != nil { + return fmt.Errorf("error migrating database compression version: %w", + err) + } + + // Move the database UTXO set version from the legacy bucket to the new + // key. + oldKey = prefixedKey(bucketId, utxoDbInfoUtxoVerKeyNameV1) + err = moveKey(tx, oldKey, utxoDbInfoUtxoVerKeyNew) + if err != nil { + return fmt.Errorf("error migrating UTXO set version: %w", err) + } + + // Move the database creation date from the legacy bucket to the new key. + oldKey = prefixedKey(bucketId, utxoDbInfoCreatedKeyNameV1) + err = moveKey(tx, oldKey, utxoDbInfoCreatedKeyNew) + if err != nil { + return fmt.Errorf("error migrating database creation date: %w", err) + } + return nil + }) + if err != nil { + return err + } + } + + // Move the UTXO set state from the legacy bucket to the new key. + err = utxoBackend.Update(func(tx UtxoBackendTx) error { + oldKey := prefixedKey(metadataLegacyBucketID, utxoSetStateKeyNameV1) + err = moveKey(tx, oldKey, utxoSetStateKeyNew) + if err != nil { + return fmt.Errorf("error migrating UTXO set state: %w", err) + } + return nil + }) + if err != nil { + return err + } + + log.Info("Migrating UTXO database. This may take a while...") + start := time.Now() + + // Move the UTXO set from the legacy bucket to the new keys. + bucketId, err = fetchLegacyBucketId(utxoBackend.Get, utxoSetLegacyBucketName) + if err != nil { + return err + } + + // If the legacy bucket doesn't exist, return as there is nothing to do. + if bucketId == nil { + return nil + } + + // doBatch contains the primary logic for migrating the UTXO set. This is + // done because attempting to migrate in a single database transaction could + // result in massive memory usage and could potentially crash on many systems + // due to ulimits. + // + // It returns whether or not all entries have been updated. + const maxEntries = 20000 + var totalMigrated uint64 + doBatch := func(tx UtxoBackendTx) (bool, error) { + // Migrate UTXO set entries so long as the max number of entries for this + // batch has not been exceeded. + var logProgress bool + var numMigrated uint32 + + iter := tx.NewIterator(bucketId) + defer iter.Release() + for iter.Next() { + // Reset err on each iteration. + err = nil + + if interruptRequested(ctx) { + logProgress = true + err = errInterruptRequested + break + } + + if numMigrated >= maxEntries { + logProgress = true + err = errBatchFinished + break + } + + // Move the UTXO set entry to the new key. + oldKey := iter.Key() + newKey := prefixedKey(utxoPrefixUtxoSetV3, oldKey[len(bucketId):]) + err = moveKey(tx, oldKey, newKey) + if err != nil { + return false, fmt.Errorf("error migrating UTXO set entry: %w", err) + } + + numMigrated++ + } + if iterErr := iter.Error(); iterErr != nil { + return false, convertLdbErr(iterErr, iterErr.Error()) + } + isFullyDone := err == nil + if (isFullyDone || logProgress) && numMigrated > 0 { + totalMigrated += uint64(numMigrated) + log.Infof("Migrated %d entries (%d total)", numMigrated, + totalMigrated) + } + return isFullyDone, err + } + + // Migrate all entries in batches for the reasons mentioned above. + if err := utxoBackendBatchedUpdate(ctx, utxoBackend, doBatch); err != nil { + return err + } + + elapsed := time.Since(start).Round(time.Millisecond) + log.Infof("Done migrating UTXO database. Total entries: %d in %v", + totalMigrated, elapsed) + + return nil +} + +// dbPutUtxoBackendInfoV1 uses an existing UTXO backend transaction to store the +// V1 UTXO backend information. +func dbPutUtxoBackendInfoV1(tx UtxoBackendTx, info *UtxoBackendInfo) error { + // V1 database info keys. + utxoDbInfoVersionKeyNameV1 := []byte("version") + utxoDbInfoCompVerKeyNameV1 := []byte("compver") + utxoDbInfoUtxoVerKeyNameV1 := []byte("utxover") + utxoDbInfoCreatedKeyNameV1 := []byte("created") + + // V1 bucket info. + bucketNameV1 := []byte("dbinfo") + bucketIdV1, err := fetchLegacyBucketId(tx.Get, bucketNameV1) + if err != nil { + return err + } + + // uint32Bytes is a helper function to convert a uint32 to a byte slice + // using the byte order specified by the database namespace. + byteOrder := binary.LittleEndian + uint32Bytes := func(ui32 uint32) []byte { + var b [4]byte + byteOrder.PutUint32(b[:], ui32) + return b[:] + } + + // uint64Bytes is a helper function to convert a uint64 to a byte slice + // using the byte order specified by the database namespace. + uint64Bytes := func(ui64 uint64) []byte { + var b [8]byte + byteOrder.PutUint64(b[:], ui64) + return b[:] + } + + // Store the database version. + verKey := prefixedKey(bucketIdV1, utxoDbInfoVersionKeyNameV1) + err = tx.Put(verKey, uint32Bytes(info.version)) + if err != nil { + return err + } + + // Store the compression version. + compVerKey := prefixedKey(bucketIdV1, utxoDbInfoCompVerKeyNameV1) + err = tx.Put(compVerKey, uint32Bytes(info.compVer)) + if err != nil { + return err + } + + // Store the UTXO set version. + utxoVerKey := prefixedKey(bucketIdV1, utxoDbInfoUtxoVerKeyNameV1) + err = tx.Put(utxoVerKey, uint32Bytes(info.utxoVer)) + if err != nil { + return err + } + + // Store the database creation date. + createdKey := prefixedKey(bucketIdV1, utxoDbInfoCreatedKeyNameV1) + return tx.Put(createdKey, uint64Bytes(uint64(info.created.Unix()))) +} + +// upgradeUtxoDbToVersion2 upgrades a UTXO database from version 1 to version 2. +func upgradeUtxoDbToVersion2(ctx context.Context, utxoBackend UtxoBackend) error { + if interruptRequested(ctx) { + return errInterruptRequested + } + + log.Info("Upgrading UTXO database to version 2...") + start := time.Now() + + // Migrate the UTXO data to use simple key prefixes rather than buckets. + err := migrateUtxoDbBuckets(ctx, utxoBackend) + if err != nil { + return err + } + + // Fetch the backend versioning info. + utxoDbInfo, err := utxoBackend.FetchInfo() + if err != nil { + return err + } + + // Update and persist the UTXO database version. + utxoDbInfo.version = 2 + err = utxoBackend.PutInfo(utxoDbInfo) + if err != nil { + return err + } + + // Update and persist the UTXO database version in the legacy V1 bucket. + // This allows older versions to identify that a newer database version exists + // in the case of a downgrade. + err = utxoBackend.Update(func(tx UtxoBackendTx) error { + return dbPutUtxoBackendInfoV1(tx, utxoDbInfo) + }) + if err != nil { + return err + } + + elapsed := time.Since(start).Round(time.Millisecond) + log.Infof("Done upgrading database in %v.", elapsed) + return nil +} + // checkDBTooOldToUpgrade returns an ErrDBTooOldToUpgrade error if the provided // database version can no longer be upgraded due to being too old. func checkDBTooOldToUpgrade(dbVersion uint32) error { @@ -3509,7 +3895,7 @@ func upgradeSpendJournal(ctx context.Context, b *BlockChain) error { // backend info housed in the passed utxo backend instance will be updated with // the latest versions. func upgradeUtxoDb(ctx context.Context, db database.DB, - utxoBackend *LevelDbUtxoBackend) error { + utxoBackend UtxoBackend) error { // Fetch the backend versioning info. utxoDbInfo, err := utxoBackend.FetchInfo() @@ -3517,6 +3903,13 @@ func upgradeUtxoDb(ctx context.Context, db database.DB, return err } + // Update to a version 2 UTXO database as needed. + if utxoDbInfo.version == 1 { + if err := upgradeUtxoDbToVersion2(ctx, utxoBackend); err != nil { + return err + } + } + // Update to a version 3 utxo set as needed. if utxoDbInfo.utxoVer == 2 { if err := upgradeUtxoSetToVersion3(ctx, db, utxoBackend); err != nil { @@ -3536,7 +3929,7 @@ func upgradeUtxoDb(ctx context.Context, db database.DB, return nil }) if blockDbUtxoSetExists { - err := separateUtxoDatabase(ctx, db, utxoBackend.db) + err := separateUtxoDatabase(ctx, db, utxoBackend) if err != nil { return err } diff --git a/blockchain/utxobackend.go b/blockchain/utxobackend.go index d1ac0e4d..7246c496 100644 --- a/blockchain/utxobackend.go +++ b/blockchain/utxobackend.go @@ -15,57 +15,134 @@ import ( "github.com/decred/dcrd/blockchain/standalone/v2" "github.com/decred/dcrd/chaincfg/chainhash" "github.com/decred/dcrd/chaincfg/v3" - "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/wire" "github.com/syndtr/goleveldb/leveldb" ldberrors "github.com/syndtr/goleveldb/leveldb/errors" + "github.com/syndtr/goleveldb/leveldb/filter" + "github.com/syndtr/goleveldb/leveldb/opt" + "github.com/syndtr/goleveldb/leveldb/util" ) const ( // currentUtxoDatabaseVersion indicates the current UTXO database version. - currentUtxoDatabaseVersion = 1 + currentUtxoDatabaseVersion = 2 - // currentUtxoSetVersion indicates the current UTXO set database version. - currentUtxoSetVersion = 3 + // utxoDbDir is the directory for the UTXO database. + utxoDbDir = "utxodb" - // utxoDbName is the UTXO database name. - utxoDbName = "utxodb" - - // utxoDbDefaultDriver is the default driver to use for the UTXO database. - utxoDbDefaultDriver = "ffldb" + // utxoLdbName is the name of UTXO leveldb database. It is itself within the + // utxoDbDir directory. + utxoLdbName = "metadata" ) -var ( - // utxoDbInfoBucketName is the name of the database bucket used to house - // global versioning and date information for the UTXO database. - utxoDbInfoBucketName = []byte("dbinfo") +// ----------------------------------------------------------------------------- +// utxoKeySet represents a top level key set in the UTXO backend. All keys in +// the UTXO backend start with a serialized prefix consisting of the key set +// and version of that key set as follows: +// +// +// +// Key Value Size Description +// key set uint8 1 byte The key set identifier, as defined below +// version uint8 1 byte The version of the key set +// +// ----------------------------------------------------------------------------- +type utxoKeySet uint8 +// These constants define the available UTXO backend key sets. +const ( + utxoKeySetDbInfo utxoKeySet = iota + 1 // 1 + utxoKeySetUtxoState // 2 + utxoKeySetUtxoSet // 3 +) + +// utxoKeySetNoVersion defines the value to be used for the version of key sets +// where versioning does not apply. +const utxoKeySetNoVersion = 0 + +// utxoKeySetVersions defines the current version for each UTXO backend key +// set. +var utxoKeySetVersions = map[utxoKeySet]uint8{ + // Note: The database info key set must remain at fixed keys so that older + // software can properly load the database versioning info, detect newer + // versions, and throw an error. + utxoKeySetDbInfo: utxoKeySetNoVersion, + utxoKeySetUtxoState: 1, + utxoKeySetUtxoSet: 3, +} + +// These variables define the serialized prefix for each key set and associated +// version. +var ( + // utxoPrefixDbInfo is the prefix for all keys in the database info + // key set. + utxoPrefixDbInfo = []byte{byte(utxoKeySetDbInfo), + utxoKeySetVersions[utxoKeySetDbInfo]} + + // utxoPrefixUtxoState is the prefix for all keys in the UTXO state key set. + utxoPrefixUtxoState = []byte{byte(utxoKeySetUtxoState), + utxoKeySetVersions[utxoKeySetUtxoState]} + + // utxoPrefixUtxoSet is the prefix for all keys in the UTXO set key set. + utxoPrefixUtxoSet = []byte{byte(utxoKeySetUtxoSet), + utxoKeySetVersions[utxoKeySetUtxoSet]} +) + +// prefixedKey returns a new byte slice that consists of the provided prefix +// appended with the provided key. +func prefixedKey(prefix []byte, key []byte) []byte { + lenPrefix := len(prefix) + prefixedKey := make([]byte, lenPrefix+len(key)) + _ = copy(prefixedKey, prefix) + _ = copy(prefixedKey[lenPrefix:], key) + return prefixedKey +} + +// These variables define keys that are part of the database info key set. +var ( // utxoDbInfoVersionKeyName is the name of the database key used to house the - // database version. It is itself under the utxoDbInfoBucketName bucket. + // database version. It is itself under utxoPrefixDbInfo. utxoDbInfoVersionKeyName = []byte("version") - // utxoDbInfoCompressionVerKeyName is the name of the database key used to - // house the database compression version. It is itself under the - // utxoDbInfoBucketName bucket. - utxoDbInfoCompressionVerKeyName = []byte("compver") + // utxoDbInfoCompVerKeyName is the name of the database key used to house the + // database compression version. It is itself under utxoPrefixDbInfo. + utxoDbInfoCompVerKeyName = []byte("compver") // utxoDbInfoUtxoVerKeyName is the name of the database key used to house the - // database UTXO set version. It is itself under the utxoDbInfoBucketName - // bucket. + // database UTXO set version. It is itself under utxoPrefixDbInfo. utxoDbInfoUtxoVerKeyName = []byte("utxover") - // utxoDbInfoCreatedKeyName is the name of the database key used to house - // the date the database was created. It is itself under the - // utxoDbInfoBucketName bucket. + // utxoDbInfoCreatedKeyName is the name of the database key used to house the + // date the database was created. It is itself under utxoPrefixDbInfo. utxoDbInfoCreatedKeyName = []byte("created") - // utxoSetBucketName is the name of the db bucket used to house the unspent - // transaction output set. - utxoSetBucketName = []byte("utxosetv3") + // utxoDbInfoVersionKey is the database key used to house the database + // version. + utxoDbInfoVersionKey = prefixedKey(utxoPrefixDbInfo, utxoDbInfoVersionKeyName) - // utxoSetStateKeyName is the name of the database key used to house the - // state of the unspent transaction output set. + // utxoDbInfoCompVerKey is the database key used to house the database + // compression version. + utxoDbInfoCompVerKey = prefixedKey(utxoPrefixDbInfo, utxoDbInfoCompVerKeyName) + + // utxoDbInfoUtxoVerKey is the database key used to house the database UTXO + // set version. + utxoDbInfoUtxoVerKey = prefixedKey(utxoPrefixDbInfo, utxoDbInfoUtxoVerKeyName) + + // utxoDbInfoCreatedKey is the database key used to house the date the + // database was created. + utxoDbInfoCreatedKey = prefixedKey(utxoPrefixDbInfo, utxoDbInfoCreatedKeyName) +) + +// These variables define keys that are part of the UTXO state key set. +var ( + // utxoSetStateKeyName is the name of the database key used to house the state + // of the unspent transaction output set. It is itself under + // utxoPrefixUtxoState. utxoSetStateKeyName = []byte("utxosetstate") + + // utxoSetStateKey is the database key used to house the state of the unspent + // transaction output set. + utxoSetStateKey = prefixedKey(utxoPrefixUtxoState, utxoSetStateKeyName) ) // ----------------------------------------------------------------------------- @@ -119,9 +196,31 @@ type UtxoBackend interface { // FetchStats returns statistics on the current UTXO set. FetchStats() (*UtxoStats, error) + // Get returns the value for the given key. It returns nil if the key does + // not exist. An empty slice is returned for keys that exist but have no + // value assigned. + // + // The returned slice is safe to modify. Additionally, it is safe to modify + // the slice passed as an argument after Get returns. + Get(key []byte) ([]byte, error) + // InitInfo loads (or creates if necessary) the UTXO backend info. InitInfo(blockDBVersion uint32) error + // NewIterator returns an iterator over the key/value pairs in the UTXO + // backend. The returned iterator is NOT safe for concurrent use, but it is + // safe to use multiple iterators concurrently, with each in a dedicated + // goroutine. + // + // The prefix parameter allows for slicing the iterator to only contain keys + // with the given prefix. A nil prefix is treated as a key BEFORE all keys. + // + // NOTE: The contents of any slice returned by the iterator should NOT be + // modified unless noted otherwise. + // + // The iterator must be released after use, by calling the Release method. + NewIterator(prefix []byte) UtxoBackendIterator + // PutInfo sets the versioning and creation information for the UTXO backend. PutInfo(info *UtxoBackendInfo) error @@ -129,6 +228,13 @@ type UtxoBackend interface { // map along with the current state. PutUtxos(utxos map[wire.OutPoint]*UtxoEntry, state *UtxoSetState) error + // Update invokes the passed function in the context of a UTXO Backend + // transaction. Any errors returned from the user-supplied function will + // cause the transaction to be rolled back and are returned from this + // function. Otherwise, the transaction is committed when the user-supplied + // function returns a nil error. + Update(fn func(tx UtxoBackendTx) error) error + // Upgrade upgrades the UTXO backend by applying all possible upgrades // iteratively as needed. Upgrade(ctx context.Context, b *BlockChain) error @@ -139,7 +245,7 @@ type UtxoBackend interface { type LevelDbUtxoBackend struct { // db is the database that contains the UTXO set. It is set when the instance // is created and is not changed afterward. - db database.DB + db *leveldb.DB } // Ensure LevelDbUtxoBackend implements the UtxoBackend interface. @@ -203,43 +309,55 @@ func removeRegressionDB(net wire.CurrencyNet, dbPath string) error { return nil } +// fileExists reports whether the named file or directory exists. +func fileExists(name string) bool { + if _, err := os.Stat(name); err != nil { + if os.IsNotExist(err) { + return false + } + } + return true +} + // LoadUtxoDB loads (or creates when needed) the UTXO database and returns a // handle to it. It also contains additional logic such as ensuring the // regression test database is clean when in regression test mode. -func LoadUtxoDB(params *chaincfg.Params, dataDir string) (database.DB, error) { - // Set the database path based on the data directory and database name. - dbPath := filepath.Join(dataDir, utxoDbName) +func LoadUtxoDB(params *chaincfg.Params, dataDir string) (*leveldb.DB, error) { + // Set the database path based on the data directory, UTXO database directory, + // and database name. + dbDir := filepath.Join(dataDir, utxoDbDir) + dbPath := filepath.Join(dbDir, utxoLdbName) // The regression test is special in that it needs a clean database for each // run, so remove it now if it already exists. - removeRegressionDB(params.Net, dbPath) + _ = removeRegressionDB(params.Net, dbPath) - // createDB is a convenience func that creates the database with the type and - // network specified in the config at the path determined above while also - // creating any intermediate directories in the configured data directory path - // as needed. - createDB := func() (database.DB, error) { - // Create the data dir if it does not exist. - err := os.MkdirAll(dataDir, 0700) - if err != nil { - return nil, err - } - return database.Create(utxoDbDefaultDriver, dbPath, params.Net) + // Ensure the full path to the database exists. + dbExists := fileExists(dbPath) + if !dbExists { + // The error can be ignored here since the call to leveldb.OpenFile will + // fail if the directory couldn't be created. + // + // NOTE: It is important that os.MkdirAll is only called if the database + // does not exist. The documentation states that os.MidirAll does nothing + // if the directory already exists. However, this has proven not to be the + // case on some less supported OSes and can lead to creating new directories + // with the wrong permissions or otherwise lead to hard to diagnose issues. + _ = os.MkdirAll(dbDir, 0700) } - // Open the existing database or create a new one as needed. + // Open the database (will create it if needed). log.Infof("Loading UTXO database from '%s'", dbPath) - db, err := database.Open(utxoDbDefaultDriver, dbPath, params.Net) + opts := opt.Options{ + ErrorIfExist: !dbExists, + Strict: opt.DefaultStrict, + Compression: opt.NoCompression, + Filter: filter.NewBloomFilter(10), + } + db, err := leveldb.OpenFile(dbPath, &opts) if err != nil { - // Return the error if it's not because the database doesn't exist. - if !errors.Is(err, database.ErrDbDoesNotExist) { - return nil, err - } - - db, err = createDB() - if err != nil { - return nil, err - } + str := fmt.Sprintf("failed to open UTXO database: %v", err) + return nil, convertLdbErr(err, str) } log.Info("UTXO database loaded") @@ -249,26 +367,94 @@ func LoadUtxoDB(params *chaincfg.Params, dataDir string) (database.DB, error) { // NewLevelDbUtxoBackend returns a new LevelDbUtxoBackend instance using the // provided database. -func NewLevelDbUtxoBackend(db database.DB) *LevelDbUtxoBackend { +func NewLevelDbUtxoBackend(db *leveldb.DB) *LevelDbUtxoBackend { return &LevelDbUtxoBackend{ db: db, } } -// dbFetchUtxoEntry uses an existing database transaction to fetch the specified -// transaction output from the utxo set. +// Get gets the value for the given key from the leveldb database. It +// returns nil for both the value and the error if the database does not +// contain the key. +// +// It is safe to modify the contents of the returned slice, and it is safe to +// modify the contents of the argument after Get returns. +func (l *LevelDbUtxoBackend) Get(key []byte) ([]byte, error) { + serialized, err := l.db.Get(key, nil) + if err != nil { + if errors.Is(err, leveldb.ErrNotFound) { + return nil, nil + } + str := fmt.Sprintf("failed to get key %x from leveldb", key) + return nil, convertLdbErr(err, str) + } + return serialized, nil +} + +// Update invokes the passed function in the context of a UTXO Backend +// transaction. Any errors returned from the user-supplied function will cause +// the transaction to be rolled back and are returned from this function. +// Otherwise, the transaction is committed when the user-supplied function +// returns a nil error. +func (l *LevelDbUtxoBackend) Update(fn func(tx UtxoBackendTx) error) error { + // Start a leveldb transaction. + // + // Note: A leveldb.Transaction is used rather than a leveldb.Batch because + // it uses significantly less memory when atomically updating a large amount + // of data. Due to the UtxoCache only flushing to the UtxoBackend + // periodically, the UtxoBackend is almost always updating a large amount of + // data at once, and thus leveldb transactions are used by default over + // batches. + ldbTx, err := l.db.OpenTransaction() + if err != nil { + return convertLdbErr(err, "failed to open leveldb transaction") + } + + if err := fn(&levelDbUtxoBackendTx{ldbTx}); err != nil { + ldbTx.Discard() + return err + } + + // Commit the leveldb transaction. + if err := ldbTx.Commit(); err != nil { + ldbTx.Discard() + return convertLdbErr(err, "failed to commit leveldb transaction") + } + return nil +} + +// NewIterator returns an iterator over the key/value pairs in the UTXO backend. +// The returned iterator is NOT safe for concurrent use, but it is safe to use +// multiple iterators concurrently, with each in a dedicated goroutine. +// +// The prefix parameter allows for slicing the iterator to only contain keys +// with the given prefix. A nil prefix is treated as a key BEFORE all keys. +// +// NOTE: The contents of any slice returned by the iterator should NOT be +// modified unless noted otherwise. +// +// The iterator must be released after use, by calling the Release method. +func (l *LevelDbUtxoBackend) NewIterator(prefix []byte) UtxoBackendIterator { + var slice *util.Range + if prefix != nil { + slice = util.BytesPrefix(prefix) + } + return l.db.NewIterator(slice, nil) +} + +// dbFetchUtxoEntry fetches the specified transaction output from the utxo set. // // When there is no entry for the provided output, nil will be returned for both // the entry and the error. -func (l *LevelDbUtxoBackend) dbFetchUtxoEntry(dbTx database.Tx, - outpoint wire.OutPoint) (*UtxoEntry, error) { - +func (l *LevelDbUtxoBackend) dbFetchUtxoEntry(outpoint wire.OutPoint) (*UtxoEntry, error) { // Fetch the unspent transaction output information for the passed transaction // output. Return now when there is no entry. key := outpointKey(outpoint) - utxoBucket := dbTx.Metadata().Bucket(utxoSetBucketName) - serializedUtxo := utxoBucket.Get(*key) + serializedUtxo, err := l.Get(*key) recycleOutpointKey(key) + if err != nil { + return nil, err + } if serializedUtxo == nil { return nil, nil } @@ -283,11 +469,11 @@ func (l *LevelDbUtxoBackend) dbFetchUtxoEntry(dbTx database.Tx, // Deserialize the utxo entry and return it. entry, err := deserializeUtxoEntry(serializedUtxo, outpoint.Index) if err != nil { - // Ensure any deserialization errors are returned as database corruption + // Ensure any deserialization errors are returned as UTXO backend corruption // errors. if isDeserializeErr(err) { str := fmt.Sprintf("corrupt utxo entry for %v: %v", outpoint, err) - return nil, makeDbErr(database.ErrCorruption, str) + return nil, contextError(ErrUtxoBackendCorruption, str) } return nil, err @@ -307,35 +493,20 @@ func (l *LevelDbUtxoBackend) FetchEntry(outpoint wire.OutPoint) (*UtxoEntry, err // will result in nil entries in the view. This is intentionally done // so other code can use the presence of an entry in the view as a way // to unnecessarily avoid attempting to reload it from the database. - var entry *UtxoEntry - err := l.db.View(func(dbTx database.Tx) error { - var err error - entry, err = l.dbFetchUtxoEntry(dbTx, outpoint) - return err - }) - if err != nil { - return nil, err - } - - return entry, nil + return l.dbFetchUtxoEntry(outpoint) } // FetchState returns the current state of the UTXO set. func (l *LevelDbUtxoBackend) FetchState() (*UtxoSetState, error) { // Fetch the utxo set state from the database. - var serialized []byte - err := l.db.View(func(dbTx database.Tx) error { - // Fetch the serialized utxo set state from the database. - serialized = dbTx.Metadata().Get(utxoSetStateKeyName) - return nil - }) + serialized, err := l.Get(utxoSetStateKey) if err != nil { return nil, err } // Return nil if the utxo set state does not exist in the database. This - // should only be the case when starting from a fresh database or a database - // that has not been run with the utxo cache yet. + // should only be the case when starting from a fresh database or a + // database that has not been run with the utxo cache yet. if serialized == nil { return nil, nil } @@ -344,25 +515,24 @@ func (l *LevelDbUtxoBackend) FetchState() (*UtxoSetState, error) { return deserializeUtxoSetState(serialized) } -// dbFetchUxtoStats fetches statistics on the current unspent transaction output -// set. -func (l *LevelDbUtxoBackend) dbFetchUtxoStats(dbTx database.Tx) (*UtxoStats, error) { - utxoBucket := dbTx.Metadata().Bucket(utxoSetBucketName) - +// FetchStats returns statistics on the current UTXO set. +func (l *LevelDbUtxoBackend) FetchStats() (*UtxoStats, error) { var stats UtxoStats transactions := make(map[chainhash.Hash]struct{}) leaves := make([]chainhash.Hash, 0) - cursor := utxoBucket.Cursor() + iter := l.NewIterator(utxoPrefixUtxoSet) + defer iter.Release() - for ok := cursor.First(); ok; ok = cursor.Next() { - key := cursor.Key() + for iter.Next() { + key := iter.Key() var outpoint wire.OutPoint err := decodeOutpointKey(key, &outpoint) if err != nil { - return nil, err + str := fmt.Sprintf("corrupt outpoint for key %x: %v", key, err) + return nil, contextError(ErrUtxoBackendCorruption, str) } - serializedUtxo := cursor.Value() + serializedUtxo := iter.Value() entrySize := len(serializedUtxo) // A non-nil zero-length entry means there is an entry in the database for a @@ -381,11 +551,11 @@ func (l *LevelDbUtxoBackend) dbFetchUtxoStats(dbTx database.Tx) (*UtxoStats, err // Deserialize the utxo entry. entry, err := deserializeUtxoEntry(serializedUtxo, outpoint.Index) if err != nil { - // Ensure any deserialization errors are returned as database corruption - // errors. + // Ensure any deserialization errors are returned as UTXO backend + // corruption errors. if isDeserializeErr(err) { str := fmt.Sprintf("corrupt utxo entry for %v: %v", outpoint, err) - return nil, makeDbErr(database.ErrCorruption, str) + return nil, contextError(ErrUtxoBackendCorruption, str) } return nil, err @@ -393,6 +563,9 @@ func (l *LevelDbUtxoBackend) dbFetchUtxoStats(dbTx database.Tx) (*UtxoStats, err stats.Total += entry.amount } + if err := iter.Error(); err != nil { + return nil, convertLdbErr(err, err.Error()) + } stats.SerializedHash = standalone.CalcMerkleRootInPlace(leaves) stats.Transactions = int64(len(transactions)) @@ -400,24 +573,9 @@ func (l *LevelDbUtxoBackend) dbFetchUtxoStats(dbTx database.Tx) (*UtxoStats, err return &stats, nil } -// FetchStats returns statistics on the current UTXO set. -func (l *LevelDbUtxoBackend) FetchStats() (*UtxoStats, error) { - var stats *UtxoStats - err := l.db.View(func(dbTx database.Tx) error { - var err error - stats, err = l.dbFetchUtxoStats(dbTx) - return err - }) - if err != nil { - return nil, err - } - - return stats, nil -} - -// dbPutUtxoBackendInfo uses an existing database transaction to store the +// dbPutUtxoBackendInfo uses an existing UTXO backend transaction to store the // backend information. -func (l *LevelDbUtxoBackend) dbPutUtxoBackendInfo(dbTx database.Tx, +func (l *LevelDbUtxoBackend) dbPutUtxoBackendInfo(tx UtxoBackendTx, info *UtxoBackendInfo) error { // uint32Bytes is a helper function to convert a uint32 to a byte slice @@ -437,65 +595,86 @@ func (l *LevelDbUtxoBackend) dbPutUtxoBackendInfo(dbTx database.Tx, } // Store the database version. - meta := dbTx.Metadata() - bucket := meta.Bucket(utxoDbInfoBucketName) - err := bucket.Put(utxoDbInfoVersionKeyName, uint32Bytes(info.version)) + err := tx.Put(utxoDbInfoVersionKey, uint32Bytes(info.version)) if err != nil { return err } // Store the compression version. - err = bucket.Put(utxoDbInfoCompressionVerKeyName, uint32Bytes(info.compVer)) + err = tx.Put(utxoDbInfoCompVerKey, uint32Bytes(info.compVer)) if err != nil { return err } // Store the UTXO set version. - err = bucket.Put(utxoDbInfoUtxoVerKeyName, uint32Bytes(info.utxoVer)) + err = tx.Put(utxoDbInfoUtxoVerKey, uint32Bytes(info.utxoVer)) if err != nil { return err } // Store the database creation date. - return bucket.Put(utxoDbInfoCreatedKeyName, + return tx.Put(utxoDbInfoCreatedKey, uint64Bytes(uint64(info.created.Unix()))) } -// dbFetchUtxoBackendInfo uses an existing database transaction to fetch the -// backend versioning and creation information. -func (l *LevelDbUtxoBackend) dbFetchUtxoBackendInfo(dbTx database.Tx) *UtxoBackendInfo { - meta := dbTx.Metadata() - bucket := meta.Bucket(utxoDbInfoBucketName) - - // Uninitialized state. - if bucket == nil { - return nil - } - +// dbFetchUtxoBackendInfo fetches the backend versioning and creation +// information. +func (l *LevelDbUtxoBackend) dbFetchUtxoBackendInfo() (*UtxoBackendInfo, error) { // Load the database version. - var version uint32 - versionBytes := bucket.Get(utxoDbInfoVersionKeyName) - if versionBytes != nil { - version = byteOrder.Uint32(versionBytes) + prefix := utxoPrefixDbInfo + versionBytes, err := l.Get(utxoDbInfoVersionKey) + if err != nil { + return nil, err } + if versionBytes == nil { + // If the database info was not found, attempt to find it in the legacy + // bucket. + dbInfoLegacyBucketName := []byte("dbinfo") + dbInfoBucketId, err := fetchLegacyBucketId(l.Get, dbInfoLegacyBucketName) + if err != nil { + return nil, err + } + prefix = dbInfoBucketId + versionBytes, err = l.Get(prefixedKey(prefix, utxoDbInfoVersionKeyName)) + if err != nil { + return nil, err + } + if versionBytes == nil { + // Uninitialized state. + return nil, nil + } + } + version := byteOrder.Uint32(versionBytes) // Load the database compression version. var compVer uint32 - compVerBytes := bucket.Get(utxoDbInfoCompressionVerKeyName) + compVerKey := prefixedKey(prefix, utxoDbInfoCompVerKeyName) + compVerBytes, err := l.Get(compVerKey) + if err != nil { + return nil, err + } if compVerBytes != nil { compVer = byteOrder.Uint32(compVerBytes) } // Load the database UTXO set version. var utxoVer uint32 - utxoVerBytes := bucket.Get(utxoDbInfoUtxoVerKeyName) + utxoVerKey := prefixedKey(prefix, utxoDbInfoUtxoVerKeyName) + utxoVerBytes, err := l.Get(utxoVerKey) + if err != nil { + return nil, err + } if utxoVerBytes != nil { utxoVer = byteOrder.Uint32(utxoVerBytes) } // Load the database creation date. var created time.Time - createdBytes := bucket.Get(utxoDbInfoCreatedKeyName) + createdKey := prefixedKey(prefix, utxoDbInfoCreatedKeyName) + createdBytes, err := l.Get(createdKey) + if err != nil { + return nil, err + } if createdBytes != nil { ts := byteOrder.Uint64(createdBytes) created = time.Unix(int64(ts), 0) @@ -506,107 +685,63 @@ func (l *LevelDbUtxoBackend) dbFetchUtxoBackendInfo(dbTx database.Tx) *UtxoBacke compVer: compVer, utxoVer: utxoVer, created: created, - } + }, nil } // createUtxoBackendInfo initializes the UTXO backend info. It must only be // called on an uninitialized backend. func (l *LevelDbUtxoBackend) createUtxoBackendInfo(blockDBVersion uint32) error { - // Create the initial UTXO database state. - err := l.db.Update(func(dbTx database.Tx) error { - meta := dbTx.Metadata() + // Initialize the UTXO set version. If the block database version is before + // version 9, then initialize the UTXO set version based on the block + // database version since that is what tracked the UTXO set version at that + // point in time. + utxoVer := uint32(utxoKeySetVersions[utxoKeySetUtxoSet]) + if blockDBVersion >= 7 && blockDBVersion < 9 { + utxoVer = 2 + } else if blockDBVersion < 7 { + utxoVer = 1 + } - // Create the bucket that houses information about the database's creation - // and version. - _, err := meta.CreateBucketIfNotExists(utxoDbInfoBucketName) - if err != nil { - return err - } - - // Initialize the UTXO set version. If the block database version is before - // version 9, then initialize the UTXO set version based on the block - // database version since that is what tracked the UTXO set version at that - // point in time. - utxoVer := uint32(currentUtxoSetVersion) - if blockDBVersion >= 7 && blockDBVersion < 9 { - utxoVer = 2 - } else if blockDBVersion < 7 { - utxoVer = 1 - } - - // Write the creation and version information to the database. - err = l.dbPutUtxoBackendInfo(dbTx, &UtxoBackendInfo{ + // Write the creation and version information to the database. + return l.Update(func(tx UtxoBackendTx) error { + return l.dbPutUtxoBackendInfo(tx, &UtxoBackendInfo{ version: currentUtxoDatabaseVersion, compVer: currentCompressionVersion, utxoVer: utxoVer, created: time.Now(), }) - if err != nil { - return err - } - - // Create the bucket that houses the UTXO set. - _, err = meta.CreateBucketIfNotExists(utxoSetBucketName) - if err != nil { - return err - } - - return nil }) - return err } // FetchInfo returns versioning and creation information for the backend. func (l *LevelDbUtxoBackend) FetchInfo() (*UtxoBackendInfo, error) { - var backendInfo *UtxoBackendInfo - err := l.db.View(func(dbTx database.Tx) error { - backendInfo = l.dbFetchUtxoBackendInfo(dbTx) - return nil - }) - if err != nil { - return nil, err - } - - return backendInfo, nil + return l.dbFetchUtxoBackendInfo() } // InitInfo loads (or creates if necessary) the UTXO backend info. func (l *LevelDbUtxoBackend) InitInfo(blockDBVersion uint32) error { - // Determine the state of the database. - var isStateInitialized bool - err := l.db.View(func(dbTx database.Tx) error { - // Fetch the backend versioning information. - dbInfo := l.dbFetchUtxoBackendInfo(dbTx) - - // The database bucket for the versioning information is missing. - if dbInfo == nil { - return nil - } - - // Don't allow downgrades of the UTXO database. - if dbInfo.version > currentUtxoDatabaseVersion { - return fmt.Errorf("the current UTXO database is no longer compatible "+ - "with this version of the software (%d > %d)", dbInfo.version, - currentUtxoDatabaseVersion) - } - - // Don't allow downgrades of the database compression version. - if dbInfo.compVer > currentCompressionVersion { - return fmt.Errorf("the current database compression version is no "+ - "longer compatible with this version of the software (%d > %d)", - dbInfo.compVer, currentCompressionVersion) - } - - isStateInitialized = true - - return nil - }) + // Fetch the backend versioning information. + dbInfo, err := l.dbFetchUtxoBackendInfo() if err != nil { return err } + // Don't allow downgrades of the UTXO database. + if dbInfo != nil && dbInfo.version > currentUtxoDatabaseVersion { + return fmt.Errorf("the current UTXO database is no longer compatible "+ + "with this version of the software (%d > %d)", dbInfo.version, + currentUtxoDatabaseVersion) + } + + // Don't allow downgrades of the database compression version. + if dbInfo != nil && dbInfo.compVer > currentCompressionVersion { + return fmt.Errorf("the current database compression version is no "+ + "longer compatible with this version of the software (%d > %d)", + dbInfo.compVer, currentCompressionVersion) + } + // Initialize the backend if it has not already been done. - if !isStateInitialized { + if dbInfo == nil { if err := l.createUtxoBackendInfo(blockDBVersion); err != nil { return err } @@ -617,17 +752,17 @@ func (l *LevelDbUtxoBackend) InitInfo(blockDBVersion uint32) error { // PutInfo sets the versioning and creation information for the backend. func (l *LevelDbUtxoBackend) PutInfo(info *UtxoBackendInfo) error { - return l.db.Update(func(dbTx database.Tx) error { - return l.dbPutUtxoBackendInfo(dbTx, info) + return l.Update(func(tx UtxoBackendTx) error { + return l.dbPutUtxoBackendInfo(tx, info) }) } -// dbPutUtxoEntry uses an existing database transaction to update the utxo +// dbPutUtxoEntry uses an existing UTXO backend transaction to update the utxo // entry for the given outpoint based on the provided utxo entry state. In // particular, the entry is only written to the database if it is marked as // modified, and if the entry is marked as spent it is removed from the // database. -func (l *LevelDbUtxoBackend) dbPutUtxoEntry(dbTx database.Tx, +func (l *LevelDbUtxoBackend) dbPutUtxoEntry(tx UtxoBackendTx, outpoint wire.OutPoint, entry *UtxoEntry) error { // No need to update the database if the entry was not modified. @@ -636,25 +771,21 @@ func (l *LevelDbUtxoBackend) dbPutUtxoEntry(dbTx database.Tx, } // Remove the utxo entry if it is spent. - utxoBucket := dbTx.Metadata().Bucket(utxoSetBucketName) if entry.IsSpent() { key := outpointKey(outpoint) - err := utxoBucket.Delete(*key) + err := tx.Delete(*key) recycleOutpointKey(key) if err != nil { return err } - return nil } // Serialize and store the utxo entry. serialized := serializeUtxoEntry(entry) key := outpointKey(outpoint) - err := utxoBucket.Put(*key, serialized) - // NOTE: The key is intentionally not recycled here since the database - // interface contract prohibits modifications. It will be garbage collected - // normally when the database is done with it. + err := tx.Put(*key, serialized) + recycleOutpointKey(key) if err != nil { return err } @@ -670,29 +801,20 @@ func (l *LevelDbUtxoBackend) PutUtxos(utxos map[wire.OutPoint]*UtxoEntry, // Update the database with the provided entries and UTXO set state. // // It is important that the UTXO set state is always updated in the same - // database transaction as the utxo set itself so that it is always in sync. - err := l.db.Update(func(dbTx database.Tx) error { + // UTXO backend transaction as the utxo set itself so that it is always in + // sync. + return l.Update(func(tx UtxoBackendTx) error { for outpoint, entry := range utxos { // Write the entry to the database. - err := l.dbPutUtxoEntry(dbTx, outpoint, entry) + err := l.dbPutUtxoEntry(tx, outpoint, entry) if err != nil { return err } } // Update the UTXO set state in the database. - return dbTx.Metadata().Put(utxoSetStateKeyName, - serializeUtxoSetState(state)) + return tx.Put(utxoSetStateKey, serializeUtxoSetState(state)) }) - if err != nil { - return err - } - - // Flush the UTXO database to disk. This is necessary in the case that the - // UTXO set state was just initialized so that if the block database is - // flushed, and then an unclean shutdown occurs, the UTXO cache will know - // where to start from when recovering on startup. - return l.db.Flush() } // Upgrade upgrades the UTXO backend by applying all possible upgrades diff --git a/blockchain/utxobackend_test.go b/blockchain/utxobackend_test.go index daafe44b..9403a7b2 100644 --- a/blockchain/utxobackend_test.go +++ b/blockchain/utxobackend_test.go @@ -6,13 +6,10 @@ package blockchain import ( "errors" - "os" - "path/filepath" "reflect" "testing" "time" - "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/wire" "github.com/syndtr/goleveldb/leveldb" ldberrors "github.com/syndtr/goleveldb/leveldb/errors" @@ -22,28 +19,13 @@ import ( func createTestUtxoBackend(t *testing.T) *LevelDbUtxoBackend { t.Helper() - // Create a test database. - dbPath := filepath.Join(os.TempDir(), t.Name()) - _ = os.RemoveAll(dbPath) - db, err := database.Create("ffldb", dbPath, wire.MainNet) + db, teardown, err := createTestUtxoDatabase(t.Name()) if err != nil { t.Fatalf("error creating test database: %v", err) } t.Cleanup(func() { - os.RemoveAll(dbPath) + teardown() }) - t.Cleanup(func() { - db.Close() - }) - - // Create the utxo set bucket. - err = db.Update(func(dbTx database.Tx) error { - _, err := dbTx.Metadata().CreateBucketIfNotExists(utxoSetBucketName) - return err - }) - if err != nil { - t.Fatalf("error creating utxo bucket: %v", err) - } return NewLevelDbUtxoBackend(db) } @@ -159,10 +141,10 @@ func TestFetchEntryFromBackend(t *testing.T) { for _, test := range tests { // Add entries specified by the test to the test backend. - err := backend.db.Update(func(dbTx database.Tx) error { + err := backend.Update(func(tx UtxoBackendTx) error { for outpoint, serialized := range test.backendEntries { key := outpointKey(outpoint) - err := dbTx.Metadata().Bucket(utxoSetBucketName).Put(*key, serialized) + err := tx.Put(*key, serialized) if err != nil { return err } @@ -336,8 +318,8 @@ func TestFetchState(t *testing.T) { for _, test := range tests { // Update the utxo set state in the backend. if test.state != nil { - err := backend.db.Update(func(dbTx database.Tx) error { - return dbTx.Metadata().Put(utxoSetStateKeyName, + err := backend.Update(func(tx UtxoBackendTx) error { + return tx.Put(utxoSetStateKey, serializeUtxoSetState(test.state)) }) if err != nil { @@ -385,17 +367,8 @@ func TestPutInfo(t *testing.T) { for _, test := range tests { if test.backendInfo != nil { - // Create the UTXO database info bucket. - err := backend.db.Update(func(dbTx database.Tx) error { - _, err := dbTx.Metadata().CreateBucketIfNotExists(utxoDbInfoBucketName) - return err - }) - if err != nil { - t.Fatalf("%q: error creating UTXO bucket: %v", test.name, err) - } - // Update the UTXO backend info. - err = backend.PutInfo(test.backendInfo) + err := backend.PutInfo(test.backendInfo) if err != nil { t.Fatalf("%q: error putting UTXO backend info: %v", test.name, err) } diff --git a/blockchain/utxoio.go b/blockchain/utxoio.go index a2f60bd1..6fbadd55 100644 --- a/blockchain/utxoio.go +++ b/blockchain/utxoio.go @@ -68,12 +68,16 @@ var maxUint32VLQSerializeSize = serializeSizeVLQ(1<<32 - 1) // serialize as a VLQ. var maxUint8VLQSerializeSize = serializeSizeVLQ(1<<8 - 1) +// utxoSetDbPrefixSize is the number of bytes that the prefx for UTXO set +// entries takes. +var utxoSetDbPrefixSize = len(utxoPrefixUtxoSet) + // outpointKeyPool defines a concurrent safe free list of byte slices used to // provide temporary buffers for outpoint database keys. var outpointKeyPool = sync.Pool{ New: func() interface{} { - b := make([]byte, chainhash.HashSize+maxUint8VLQSerializeSize+ - maxUint32VLQSerializeSize) + b := make([]byte, utxoSetDbPrefixSize+chainhash.HashSize+ + maxUint8VLQSerializeSize+maxUint32VLQSerializeSize) return &b // Pointer to slice to avoid boxing alloc. }, } @@ -91,10 +95,12 @@ func outpointKey(outpoint wire.OutPoint) *[]byte { key := outpointKeyPool.Get().(*[]byte) tree := uint64(outpoint.Tree) idx := uint64(outpoint.Index) - *key = (*key)[:chainhash.HashSize+serializeSizeVLQ(tree) + - +serializeSizeVLQ(idx)] - copy(*key, outpoint.Hash[:]) - offset := chainhash.HashSize + *key = (*key)[:utxoSetDbPrefixSize+chainhash.HashSize+ + serializeSizeVLQ(tree)+serializeSizeVLQ(idx)] + copy(*key, utxoPrefixUtxoSet) + offset := utxoSetDbPrefixSize + copy((*key)[offset:], outpoint.Hash[:]) + offset += chainhash.HashSize offset += putVLQ((*key)[offset:], tree) putVLQ((*key)[offset:], idx) return key @@ -102,14 +108,17 @@ func outpointKey(outpoint wire.OutPoint) *[]byte { // decodeOutpointKey decodes the passed serialized key into the passed outpoint. func decodeOutpointKey(serialized []byte, outpoint *wire.OutPoint) error { - if chainhash.HashSize >= len(serialized) { + if utxoSetDbPrefixSize+chainhash.HashSize >= len(serialized) { return errDeserialize("unexpected length for serialized outpoint key") } + // Ignore the UTXO set prefix. + offset := utxoSetDbPrefixSize + // Deserialize the hash. var hash chainhash.Hash - copy(hash[:], serialized[:chainhash.HashSize]) - offset := chainhash.HashSize + copy(hash[:], serialized[offset:offset+chainhash.HashSize]) + offset += chainhash.HashSize // Deserialize the tree. tree, bytesRead := deserializeVLQ(serialized[offset:]) diff --git a/blockchain/validate_test.go b/blockchain/validate_test.go index 8a526f57..17e2fcb9 100644 --- a/blockchain/validate_test.go +++ b/blockchain/validate_test.go @@ -282,8 +282,7 @@ func TestCheckBlockHeaderContext(t *testing.T) { defer teardownDb() // Create a test UTXO database. - utxoDb, teardownUtxoDb, err := createTestDatabase(dbName+"_utxo", testDbType, - params.Net) + utxoDb, teardownUtxoDb, err := createTestUtxoDatabase(dbName + "_utxo") if err != nil { t.Fatalf("Failed to create UTXO database: %v\n", err) } diff --git a/internal/netsync/manager.go b/internal/netsync/manager.go index 3573b2be..951dfd41 100644 --- a/internal/netsync/manager.go +++ b/internal/netsync/manager.go @@ -775,7 +775,9 @@ func (m *SyncManager) handleBlockMsg(bmsg *blockMsg) { } else { log.Errorf("Failed to process block %v: %v", blockHash, err) } - if errors.Is(err, database.ErrCorruption) { + if errors.Is(err, database.ErrCorruption) || + errors.Is(err, blockchain.ErrUtxoBackendCorruption) { + log.Errorf("Critical failure: %v", err) } return diff --git a/server.go b/server.go index 34602af6..41aaa800 100644 --- a/server.go +++ b/server.go @@ -48,6 +48,7 @@ import ( "github.com/decred/dcrd/peer/v3" "github.com/decred/dcrd/txscript/v4" "github.com/decred/dcrd/wire" + "github.com/syndtr/goleveldb/leveldb" ) const ( @@ -3294,7 +3295,7 @@ func setupRPCListeners() ([]net.Listener, error) { // decred network type specified by chainParams. Use start to begin accepting // connections from peers. func newServer(ctx context.Context, listenAddrs []string, db database.DB, - utxoDb database.DB, chainParams *chaincfg.Params, + utxoDb *leveldb.DB, chainParams *chaincfg.Params, dataDir string) (*server, error) { amgr := addrmgr.New(cfg.DataDir, dcrdLookup)