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