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)