From eea386e4e918059cf89afc42d77462e6c35d43ee Mon Sep 17 00:00:00 2001 From: Dave Collins Date: Fri, 20 Sep 2019 21:30:08 -0500 Subject: [PATCH] blockchain: Implement v2 block filter storage. This modifies the chain logic to create and store version 2 block filters for all new blocks and also adds code to migrate the database to retroactively create and store the v2 filters for all historical blocks. Since this requires a database upgrade and the next release of the software will include a vote to change the consensus rules, this also takes the opportunity to unmark all blocks previously marked as having failed validation so they are eligible for validation again under what will likely become new consensus rules. This ensures clients that did not update prior to new rules activating are able to automatically recover under the new rules without having to download the entire chain again. The following is a high level overview of the changes: - Introduce a new database bucket to house v2 block filters - Make UtxoViewpoint satisfy the PrevScripter interface so it may be used as a source previous scripts when creating filters - Create and store the new filters in the db when connecting blocks - Introduce exported function named FilterByBlockHash to retrieve the new block filters so they are available to be served in the future - Implement database migration code to retroactively create the new filters for all historical blocks - Bump the chain database version to 6 - Introduce code to allow spent txout entries from the spend journal to be used as a source of previous scripts to significantly optimize the filter creation as compared to what would be required to reconstruct all the utxo views as of each block - Mark all blocks that failed validation under the current consensus rules as eligible for validation again - Export a new constant named HeaderCmtFilterIndex which indicates the header proof index for the upcoming filter header commitment --- blockchain/chain.go | 19 ++ blockchain/chainio.go | 75 +++- blockchain/error.go | 10 + blockchain/headercmt.go | 37 ++ .../internal/dbnamespace/dbnamespace.go | 4 + blockchain/upgrade.go | 319 ++++++++++++++++++ blockchain/utxoviewpoint.go | 16 + 7 files changed, 478 insertions(+), 2 deletions(-) create mode 100644 blockchain/headercmt.go diff --git a/blockchain/chain.go b/blockchain/chain.go index e4721f22..11ade983 100644 --- a/blockchain/chain.go +++ b/blockchain/chain.go @@ -17,6 +17,7 @@ import ( "github.com/decred/dcrd/chaincfg/v2" "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/dcrutil/v2" + "github.com/decred/dcrd/gcs/v2/blockcf2" "github.com/decred/dcrd/txscript/v2" "github.com/decred/dcrd/wire" ) @@ -711,6 +712,14 @@ func (b *BlockChain) connectBlock(node *blockNode, block, parent *dcrutil.Block, return err } + // This ultimately will need to be created earlier in the validation so the + // header commitment root can be validated. However, it is here for now in + // order to add support for filter storage and retrieval independently. + filter, err := blockcf2.Regular(block.MsgBlock(), view) + if err != nil { + return ruleError(ErrMissingTxOut, err.Error()) + } + // Generate a new best state snapshot that will be used to update the // database and later memory if all database updates are successful. b.stateLock.RLock() @@ -755,6 +764,12 @@ func (b *BlockChain) connectBlock(node *blockNode, block, parent *dcrutil.Block, return err } + // Insert the GCS filter for the block into the database. + err = dbPutGCSFilter(dbTx, block.Hash(), filter) + if err != nil { + return err + } + // Allow the index manager to call each of the currently active // optional indexes with the block being connected so they can // update themselves accordingly. @@ -926,6 +941,10 @@ func (b *BlockChain) disconnectBlock(node *blockNode, block, parent *dcrutil.Blo return err } + // NOTE: The GCS filter is intentionally not removed on disconnect to + // ensure that lightweight clients still have access to them if they + // happen to be on a side chain after coming back online after a reorg. + // Allow the index manager to call each of the currently active // optional indexes with the block being disconnected so they // can update themselves accordingly. diff --git a/blockchain/chainio.go b/blockchain/chainio.go index 7e113376..52c51750 100644 --- a/blockchain/chainio.go +++ b/blockchain/chainio.go @@ -18,13 +18,15 @@ import ( "github.com/decred/dcrd/chaincfg/chainhash" "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/dcrutil/v2" + "github.com/decred/dcrd/gcs/v2" + "github.com/decred/dcrd/gcs/v2/blockcf2" "github.com/decred/dcrd/wire" ) const ( // currentDatabaseVersion indicates what the current database // version is. - currentDatabaseVersion = 5 + currentDatabaseVersion = 6 // currentBlockIndexVersion indicates what the current block index // database version. @@ -1216,6 +1218,59 @@ func dbPutUtxoView(dbTx database.Tx, view *UtxoViewpoint) error { return nil } +// ----------------------------------------------------------------------------- +// The GCS filter journal consists of an entry for each block connected to the +// main chain (or has ever been connected to it) which consists of a serialized +// GCS filter. +// +// The serialized key format is: +// +// +// +// Field Type Size +// block hash chainhash.Hash chainhash.HashSize +// +// The serialized value format is: +// +// +// +// Field Type Size +// filter []byte (gcs.FilterV2) variable +// +// ----------------------------------------------------------------------------- + +// dbFetchGCSFilter fetches the GCS filter for the passed block and deserializes +// it into a slice of spent txout entries. +// +// When there is no entry for the provided hash, nil will be returned for both +// the filter and the error. +func dbFetchGCSFilter(dbTx database.Tx, blockHash *chainhash.Hash) (*gcs.FilterV2, error) { + filterBucket := dbTx.Metadata().Bucket(dbnamespace.GCSFilterBucketName) + serialized := filterBucket.Get(blockHash[:]) + if serialized == nil { + return nil, nil + } + + filter, err := gcs.FromBytesV2(blockcf2.B, blockcf2.M, serialized) + if err != nil { + return nil, database.Error{ + ErrorCode: database.ErrCorruption, + Description: fmt.Sprintf("corrupt filter for %v: %v", blockHash, + err), + } + } + + return filter, nil +} + +// dbPutGCSFilter uses an existing database transaction to update the GCS filter +// for the given block hash using the provided filter. +func dbPutGCSFilter(dbTx database.Tx, blockHash *chainhash.Hash, filter *gcs.FilterV2) error { + filterBucket := dbTx.Metadata().Bucket(dbnamespace.GCSFilterBucketName) + serialized := filter.Bytes() + return filterBucket.Put(blockHash[:], serialized) +} + // ----------------------------------------------------------------------------- // The database information contains information about the version and date // of the blockchain database. @@ -1542,7 +1597,23 @@ func (b *BlockChain) createChainState() error { } // Store the genesis block into the database. - return dbTx.StoreBlock(genesisBlock) + err = dbTx.StoreBlock(genesisBlock) + if err != nil { + return err + } + + // Create the bucket that houses the gcs filters. + _, err = meta.CreateBucket(dbnamespace.GCSFilterBucketName) + if err != nil { + return err + } + + // Store the (empty) GCS filter for the genesis block. + genesisFilter, err := blockcf2.Regular(genesisBlock.MsgBlock(), nil) + if err != nil { + return err + } + return dbPutGCSFilter(dbTx, &b.chainParams.GenesisHash, genesisFilter) }) return err } diff --git a/blockchain/error.go b/blockchain/error.go index c72a375c..756e0772 100644 --- a/blockchain/error.go +++ b/blockchain/error.go @@ -50,6 +50,16 @@ func (e DuplicateDeploymentError) Error() string { string(e)) } +// NoFilterError identifies an error that indicates a filter for a given block +// hash does not exist. +type NoFilterError string + +// Error returns the error as a human-readable string and satisfies the error +// interface. +func (e NoFilterError) Error() string { + return fmt.Sprintf("no filter available for block %s", string(e)) +} + // AssertError identifies an error that indicates an internal code consistency // issue and should be treated as a critical and unrecoverable error. type AssertError string diff --git a/blockchain/headercmt.go b/blockchain/headercmt.go new file mode 100644 index 00000000..108a194f --- /dev/null +++ b/blockchain/headercmt.go @@ -0,0 +1,37 @@ +// Copyright (c) 2019 The Decred developers +// Use of this source code is governed by an ISC +// license that can be found in the LICENSE file. + +package blockchain + +import ( + "github.com/decred/dcrd/chaincfg/chainhash" + "github.com/decred/dcrd/database/v2" + "github.com/decred/dcrd/gcs/v2" +) + +const ( + // HeaderCmtFilterIndex is the proof index for the filter header commitment. + HeaderCmtFilterIndex = 0 +) + +// FilterByBlockHash returns the version 2 GCS filter for the given block hash +// when it exists. This function returns the filters regardless of whether or +// not their associated block is part of the main chain. +// +// An error of type NoFilterError will be returned when the filter for the given +// block hash does not exist. +// +// This function is safe for concurrent access. +func (b *BlockChain) FilterByBlockHash(hash *chainhash.Hash) (*gcs.FilterV2, error) { + var filter *gcs.FilterV2 + err := b.db.View(func(dbTx database.Tx) error { + var err error + filter, err = dbFetchGCSFilter(dbTx, hash) + return err + }) + if err == nil && filter == nil { + err = NoFilterError(hash.String()) + } + return filter, err +} diff --git a/blockchain/internal/dbnamespace/dbnamespace.go b/blockchain/internal/dbnamespace/dbnamespace.go index af43e646..fe4fb859 100644 --- a/blockchain/internal/dbnamespace/dbnamespace.go +++ b/blockchain/internal/dbnamespace/dbnamespace.go @@ -52,4 +52,8 @@ var ( // block index which consists of metadata for all known blocks both in // the main chain and on side chains. BlockIndexBucketName = []byte("blockidx") + + // GCSFilterBucketName is the name of the db bucket used to house GCS + // filters. + GCSFilterBucketName = []byte("gcsfilters") ) diff --git a/blockchain/upgrade.go b/blockchain/upgrade.go index f2afaaf9..9ab109a0 100644 --- a/blockchain/upgrade.go +++ b/blockchain/upgrade.go @@ -19,6 +19,8 @@ import ( "github.com/decred/dcrd/chaincfg/v2" "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/dcrutil/v2" + "github.com/decred/dcrd/gcs/v2" + "github.com/decred/dcrd/gcs/v2/blockcf2" "github.com/decred/dcrd/wire" ) @@ -721,6 +723,312 @@ func (b *BlockChain) maybeFinishV5Upgrade() error { return nil } +// scriptSourceEntry houses a script and its associated version. +type scriptSourceEntry struct { + version uint16 + script []byte +} + +// scriptSource provides a source of transaction output scripts and their +// associated script version for given outpoints and implements the PrevScripter +// interface so it may be used in cases that require access to said scripts. +type scriptSource map[wire.OutPoint]scriptSourceEntry + +// PrevScript returns the script and script version associated with the provided +// previous outpoint along with a bool that indicates whether or not the +// requested entry exists. This ensures the caller is able to distinguish +// between missing entry and empty v0 scripts. +func (s scriptSource) PrevScript(prevOut *wire.OutPoint) (uint16, []byte, bool) { + entry, ok := s[*prevOut] + if !ok { + return 0, nil, false + } + return entry.version, entry.script, true +} + +// stxosToScriptSource uses the provided block and spent txo information to +// create a source of previous transaction scripts and versions spent by the +// block. +func stxosToScriptSource(block *dcrutil.Block, stxos []spentTxOut, compressionVersion uint32) scriptSource { + source := make(scriptSource) + + // Loop through all of the transaction inputs in the stake transaction tree + // (except for the stakebases which have no inputs) and add the scripts and + // associated script versions from the referenced txos to the script source. + // + // Note that transactions in the stake tree are spent before transactions in + // the regular tree when originally creating the spend journal entry, thus + // the spent txous need to be processed in the same order. + var stxoIdx int + for _, tx := range block.MsgBlock().STransactions { + isVote := stake.IsSSGen(tx) + for txInIdx, txIn := range tx.TxIn { + // Ignore stakebase since it has no input. + if isVote && txInIdx == 0 { + continue + } + + // Ensure the spent txout index is incremented to stay in sync with + // the transaction input. + stxo := &stxos[stxoIdx] + stxoIdx++ + + // Create an output for the referenced script and version using the + // stxo data from the spend journal if it doesn't already exist in + // the view. + prevOut := &txIn.PreviousOutPoint + source[*prevOut] = scriptSourceEntry{ + version: stxo.scriptVersion, + script: decompressScript(stxo.pkScript, compressionVersion), + } + } + } + + // Loop through all of the transaction inputs in the regular transaction + // tree (except for the coinbase which has no inputs) and add the scripts + // and associated script versions from the referenced txos to the script + // source. + for _, tx := range block.MsgBlock().Transactions[1:] { + for _, txIn := range tx.TxIn { + // Ensure the spent txout index is incremented to stay in sync with + // the transaction input. + stxo := &stxos[stxoIdx] + stxoIdx++ + + // Create an output for the referenced script and version using the + // stxo data from the spend journal if it doesn't already exist in + // the view. + prevOut := &txIn.PreviousOutPoint + source[*prevOut] = scriptSourceEntry{ + version: stxo.scriptVersion, + script: decompressScript(stxo.pkScript, compressionVersion), + } + } + } + + return source +} + +// clearFailedBlockFlags unmarks all blocks 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 activating are able to +// automatically recover under the new rules without having to download the +// entire chain again. +func clearFailedBlockFlags(index *blockIndex, interrupt <-chan struct{}) error { + for _, node := range index.index { + index.UnsetStatusFlags(node, statusValidateFailed|statusInvalidAncestor) + } + + return index.flush() +} + +// initializeGCSFilters creates and stores version 2 GCS filters for all blocks +// in the main chain. This ensures they are immediately available to clients +// and simplifies the rest of the related code since it can rely on the filters +// being available once the upgrade completes. +// +// The database is guaranteed to have a filter entry for every block in the +// main chain if this returns without failure. +func initializeGCSFilters(db database.DB, index *blockIndex, bestChain *chainView, interrupt <-chan struct{}) error { + // Hardcoded values so updates to the global values do not affect old + // upgrades. + gcsBucketName := []byte("gcsfilters") + const compressionVersion = 1 + + log.Info("Creating and storing GCS filters. This will take a while...") + start := time.Now() + + // Create the new filter bucket as needed. + err := db.Update(func(dbTx database.Tx) error { + _, err := dbTx.Metadata().CreateBucketIfNotExists(gcsBucketName) + return err + }) + if err != nil { + return err + } + + // newFilter loads the full block for the provided node from the db along + // with its spend journal information and uses it to create a v2 GCS filter. + newFilter := func(dbTx database.Tx, node *blockNode) (*gcs.FilterV2, error) { + // Load the full block from the database. + block, err := dbFetchBlockByNode(dbTx, node) + if err != nil { + return nil, err + } + + // Load all of the spent transaction output data from the database. + stxos, err := dbFetchSpendJournalEntry(dbTx, block) + if err != nil { + return nil, err + } + + // Use the combination of the block and the stxos to create a source + // of previous scripts spent by the block needed to create the + // filter. + prevScripts := stxosToScriptSource(block, stxos, compressionVersion) + + // Create the filter from the block and referenced previous output + // scripts. + filter, err := blockcf2.Regular(block.MsgBlock(), prevScripts) + if err != nil { + return nil, err + } + + return filter, nil + } + + // doBatch contains the primary logic for creating the GCS filters when + // moving from database version 5 to 6 in batches. This is done because + // attempting to create them all in a single database transaction could + // result in massive memory usage and could potentially crash on many + // systems due to ulimits. + // + // It returns the number of entries processed as well as the total number + // bytes occupied by all of the processed filters. + const maxEntries = 20000 + node := bestChain.Genesis() + doBatch := func(dbTx database.Tx) (uint64, uint64, error) { + filterBucket := dbTx.Metadata().Bucket(gcsBucketName) + if filterBucket == nil { + return 0, 0, fmt.Errorf("bucket %s does not exist", gcsBucketName) + } + + var numCreated, totalBytes uint64 + for ; node != nil; node = bestChain.Next(node) { + if numCreated >= maxEntries { + break + } + + // Create the filter from the block and referenced previous output + // scripts. + filter, err := newFilter(dbTx, node) + if err != nil { + return numCreated, totalBytes, err + } + + // Store the filter to the database. + serialized := filter.Bytes() + err = filterBucket.Put(node.hash[:], serialized) + if err != nil { + return numCreated, totalBytes, err + } + totalBytes += uint64(len(serialized)) + + numCreated++ + + if interruptRequested(interrupt) { + return numCreated, totalBytes, errInterruptRequested + } + } + + return numCreated, totalBytes, nil + } + + // Migrate all entries in batches for the reasons mentioned above. + var totalCreated, totalFilterBytes uint64 + for { + var numCreated, numFilterBytes uint64 + err := db.Update(func(dbTx database.Tx) error { + var err error + numCreated, numFilterBytes, err = doBatch(dbTx) + if err == errInterruptRequested { + // 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. + err = nil + } + return err + }) + if err != nil { + return err + } + + if interruptRequested(interrupt) { + return errInterruptRequested + } + + if numCreated == 0 { + break + } + + totalCreated += numCreated + totalFilterBytes += numFilterBytes + log.Infof("Created %d entries (%d total)", numCreated, totalCreated) + } + + elapsed := time.Since(start).Round(time.Millisecond) + log.Infof("Done creating GCS filters in %v. Total entries: %d (%d bytes)", + elapsed, totalCreated, totalFilterBytes) + return nil +} + +// upgradeToVersion6 upgrades a version 5 blockchain database to version 6. +func upgradeToVersion6(db database.DB, chainParams *chaincfg.Params, dbInfo *databaseInfo, interrupt <-chan struct{}) error { + if interruptRequested(interrupt) { + return errInterruptRequested + } + + log.Info("Upgrading database to version 6...") + start := time.Now() + + // Load the chain state and block index from the database. + bestChain := newChainView(nil) + index := newBlockIndex(db) + err := db.View(func(dbTx database.Tx) error { + // Fetch the stored best chain state from the database. + state, err := dbFetchBestState(dbTx) + if err != nil { + return err + } + + // Load all of the block index entries from the database and + // construct the block index. + err = loadBlockIndex(dbTx, &chainParams.GenesisHash, index) + if err != nil { + return err + } + + // Set the best chain to the stored best state. + tip := index.lookupNode(&state.hash) + if tip == nil { + return AssertError(fmt.Sprintf("initChainState: cannot find "+ + "chain tip %s in block index", state.hash)) + } + bestChain.SetTip(tip) + + return nil + }) + if err != nil { + return err + } + + // Unmark all blocks previously marked failed so they are eligible for + // validation again under the new consensus rules. + if err := clearFailedBlockFlags(index, interrupt); err != nil { + return err + } + + // Create and store version 2 GCS filters for all blocks in the main chain. + err = initializeGCSFilters(db, index, bestChain, interrupt) + if err != nil { + return err + } + + err = db.Update(func(dbTx database.Tx) error { + // Update and persist the updated database versions. + dbInfo.version = 6 + return dbPutDatabaseInfo(dbTx, dbInfo) + }) + if err != nil { + return err + } + + elapsed := time.Since(start).Round(time.Millisecond) + log.Infof("Done upgrading database in %v.", elapsed) + return nil +} + // upgradeDB upgrades old database versions to the newest version by applying // all possible upgrades iteratively. // @@ -757,5 +1065,16 @@ func upgradeDB(db database.DB, chainParams *chaincfg.Params, dbInfo *databaseInf } } + // Update to a version 6 database if needed. This entails unmarking all + // blocks previously marked failed so they are eligible for validation again + // under the new consensus rules and creating and storing version 2 GCS + // filters for all blocks in the main chain. + if dbInfo.version == 5 { + err := upgradeToVersion6(db, chainParams, dbInfo, interrupt) + if err != nil { + return err + } + } + return nil } diff --git a/blockchain/utxoviewpoint.go b/blockchain/utxoviewpoint.go index 03f20697..3ff5120a 100644 --- a/blockchain/utxoviewpoint.go +++ b/blockchain/utxoviewpoint.go @@ -14,6 +14,7 @@ import ( "github.com/decred/dcrd/database/v2" "github.com/decred/dcrd/dcrutil/v2" "github.com/decred/dcrd/txscript/v2" + "github.com/decred/dcrd/wire" ) // utxoOutput houses details about an individual unspent transaction output such @@ -277,6 +278,21 @@ func (view *UtxoViewpoint) LookupEntry(txHash *chainhash.Hash) *UtxoEntry { return entry } +// PrevScript returns the script and script version associated with the provided +// previous outpoint along with a bool that indicates whether or not the +// requested entry exists. This ensures the caller is able to distinguish +// between missing entry and empty v0 scripts. +func (view *UtxoViewpoint) PrevScript(prevOut *wire.OutPoint) (uint16, []byte, bool) { + entry := view.LookupEntry(&prevOut.Hash) + if entry == nil { + return 0, nil, false + } + + version := entry.ScriptVersionByIndex(prevOut.Index) + pkScript := entry.PkScriptByIndex(prevOut.Index) + return version, pkScript, true +} + // AddTxOuts adds all outputs in the passed transaction which are not provably // unspendable to the view. When the view already has entries for any of the // outputs, they are simply marked unspent. All fields will be updated for