From eeaa6d71dc7212748066c261b0eaf2b6e1a14dfd Mon Sep 17 00:00:00 2001 From: Donald Adu-Poku Date: Sun, 13 Jun 2021 18:29:15 +0000 Subject: [PATCH] indexers: async transaction index. This reworks the TransactionIndex to implement the updated Indexer interface for asynchoronous index updates. This index subscribes with no prequisites. --- blockchain/indexers/txindex.go | 182 +++++++-- blockchain/indexers/txindex_test.go | 564 ++++++++++++++++++++++++++++ 2 files changed, 721 insertions(+), 25 deletions(-) create mode 100644 blockchain/indexers/txindex_test.go diff --git a/blockchain/indexers/txindex.go b/blockchain/indexers/txindex.go index 26be5329..68148b0e 100644 --- a/blockchain/indexers/txindex.go +++ b/blockchain/indexers/txindex.go @@ -9,8 +9,10 @@ import ( "context" "errors" "fmt" + "sync" "github.com/decred/dcrd/chaincfg/chainhash" + "github.com/decred/dcrd/chaincfg/v3" "github.com/decred/dcrd/database/v3" "github.com/decred/dcrd/dcrutil/v4" "github.com/decred/dcrd/wire" @@ -322,8 +324,20 @@ func dbRemoveTxIndexEntries(dbTx database.Tx, block *dcrutil.Block) error { // TxIndex implements a transaction by hash index. That is to say, it supports // querying all transactions by their hash. type TxIndex struct { - db database.DB curBlockID uint32 + + // These fields provide access to the chain queryer and the + // database of the index. + db database.DB + chain ChainQueryer + + // These fields track the notification subscription for the index + // and its subscribers. + sub *IndexSubscription + subscribers map[chan bool]struct{} + + mtx sync.Mutex + cancel context.CancelFunc } // Ensure the TxIndex type implements the Indexer interface. @@ -334,7 +348,31 @@ var _ Indexer = (*TxIndex)(nil) // disconnecting blocks. // // This is part of the Indexer interface. -func (idx *TxIndex) Init() error { +func (idx *TxIndex) Init(ctx context.Context, chainParams *chaincfg.Params) error { + if interruptRequested(ctx) { + return errInterruptRequested + } + + // Finish any drops that were previously interrupted. + if err := finishDrop(ctx, idx); err != nil { + return err + } + + // Create the initial state for the index as needed. + if err := createIndex(idx, &chainParams.GenesisHash); err != nil { + return err + } + + // Upgrade the index as needed. + if err := upgradeIndex(ctx, idx, &chainParams.GenesisHash); err != nil { + return err + } + + // Recover the tx index and its dependents to the main chain if needed. + if err := recover(ctx, idx); err != nil { + return err + } + // Find the latest known block id field for the internal block id // index and initialize it. This is done because it's a lot more // efficient to do a single search at initialize time than it is to @@ -413,9 +451,59 @@ func (idx *TxIndex) Version() uint32 { return txIndexVersion } -// Create is invoked when the indexer manager determines the index needs -// to be created for the first time. It creates the buckets for the hash-based -// transaction index and the internal block ID indexes. +// DB returns the database of the index. +// +// This is part of the Indexer interface. +func (idx *TxIndex) DB() database.DB { + return idx.db +} + +// Queryer returns the chain queryer. +// +// This is part of the Indexer interface. +func (idx *TxIndex) Queryer() ChainQueryer { + return idx.chain +} + +// Tip returns the current tip of the index. +// +// This is part of the Indexer interface. +func (idx *TxIndex) Tip() (int64, *chainhash.Hash, error) { + return tip(idx.db, idx.Key()) +} + +// IndexSubscription returns the subscription for index updates. +// +// This is part of the Indexer interface. +func (idx *TxIndex) IndexSubscription() *IndexSubscription { + return idx.sub +} + +// Subscribers returns all client channels waiting for the next index update. +// +// This is part of the Indexer interface. +func (idx *TxIndex) Subscribers() map[chan bool]struct{} { + idx.mtx.Lock() + defer idx.mtx.Unlock() + return idx.subscribers +} + +// WaitForSync subscribes clients for the next index sync update. +// +// This is part of the Indexer interface. +func (idx *TxIndex) WaitForSync() chan bool { + c := make(chan bool) + + idx.mtx.Lock() + idx.subscribers[c] = struct{}{} + idx.mtx.Unlock() + + return c +} + +// Create is invoked when the index is created for the first time. It +// creates the buckets for the hash-based transaction index and the +// internal block ID indexes. // // This is part of the Indexer interface. func (idx *TxIndex) Create(dbTx database.Tx) error { @@ -430,12 +518,9 @@ func (idx *TxIndex) Create(dbTx database.Tx) error { return err } -// ConnectBlock is invoked by the index manager when a new block has been -// connected to the main chain. This indexer adds a hash-to-transaction mapping -// for every transaction in the passed block. -// -// This is part of the Indexer interface. -func (idx *TxIndex) ConnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, _ PrevScripter, isTreasuryEnabled bool) error { +// connectBlock adds a hash-to-transaction mapping for every transaction in +// the passed block. +func (idx *TxIndex) connectBlock(dbTx database.Tx, block, parent *dcrutil.Block, _ PrevScripter, _ bool) error { // NOTE: The fact that the block can disapprove the regular tree of the // previous block is ignored for this index because even though the // disapproved transactions no longer apply spend semantics, they still @@ -466,15 +551,14 @@ func (idx *TxIndex) ConnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, return err } idx.curBlockID = newBlockID - return nil + + // Update the current index tip. + return dbPutIndexerTip(dbTx, idx.Key(), block.Hash(), int32(block.Height())) } -// DisconnectBlock is invoked by the index manager when a block has been -// disconnected from the main chain. This indexer removes the -// hash-to-transaction mapping for every transaction in the block. -// -// This is part of the Indexer interface. -func (idx *TxIndex) DisconnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, _ PrevScripter, isTreasuryEnabled bool) error { +// disconnectBlock removes the hash-to-transaction mapping for every +// transaction in the passed block. +func (idx *TxIndex) disconnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, _ PrevScripter, _ bool) error { // NOTE: The fact that the block can disapprove the regular tree of the // previous block is ignored when disconnecting blocks because it is also // ignored when connecting the block. See the comments in ConnectBlock for @@ -491,7 +575,10 @@ func (idx *TxIndex) DisconnectBlock(dbTx database.Tx, block, parent *dcrutil.Blo return err } idx.curBlockID-- - return nil + + // Update the current index tip. + return dbPutIndexerTip(dbTx, idx.Key(), &block.MsgBlock().Header.PrevBlock, + int32(block.Height()-1)) } // Entry returns details for the provided transaction hash from the transaction @@ -513,12 +600,29 @@ func (idx *TxIndex) Entry(hash *chainhash.Hash) (*TxIndexEntry, error) { // NewTxIndex returns a new instance of an indexer that is used to create a // mapping of the hashes of all transactions in the blockchain to the respective // block, location within the block, and size of the transaction. -// -// It implements the Indexer interface which plugs into the IndexManager that in -// turn is used by the blockchain package. This allows the index to be -// seamlessly maintained along with the chain. -func NewTxIndex(db database.DB) *TxIndex { - return &TxIndex{db: db} +func NewTxIndex(subscriber *IndexSubscriber, db database.DB, chain ChainQueryer) (*TxIndex, error) { + idx := &TxIndex{ + db: db, + chain: chain, + subscribers: make(map[chan bool]struct{}), + cancel: subscriber.cancel, + } + + // The transaction index is an optional index. It has no prequisite and + // is updated asynchronously. + sub, err := subscriber.Subscribe(idx, noPrereqs) + if err != nil { + return nil, err + } + + idx.sub = sub + + err = idx.Init(subscriber.ctx, chain.ChainParams()) + if err != nil { + return nil, err + } + + return idx, nil } // dropBlockIDIndex drops the internal block id index. @@ -599,3 +703,31 @@ func DropTxIndex(ctx context.Context, db database.DB) error { func (*TxIndex) DropIndex(ctx context.Context, db database.DB) error { return DropTxIndex(ctx, db) } + +// ProcessNotification indexes the provided notification based on its +// notification type. +// +// This is part of the Indexer interface. +func (idx *TxIndex) ProcessNotification(dbTx database.Tx, ntfn *IndexNtfn) error { + switch ntfn.NtfnType { + case ConnectNtfn: + err := idx.connectBlock(dbTx, ntfn.Block, ntfn.Parent, + ntfn.PrevScripts, ntfn.IsTreasuryEnabled) + if err != nil { + return fmt.Errorf("%s: unable to connect block: %v", idx.Name(), err) + } + + case DisconnectNtfn: + err := idx.disconnectBlock(dbTx, ntfn.Block, ntfn.Parent, + ntfn.PrevScripts, ntfn.IsTreasuryEnabled) + if err != nil { + log.Errorf("%s: unable to disconnect block: %v", idx.Name(), err) + } + + default: + return fmt.Errorf("%s: unknown notification type received: %d", + idx.Name(), ntfn.NtfnType) + } + + return nil +} diff --git a/blockchain/indexers/txindex_test.go b/blockchain/indexers/txindex_test.go new file mode 100644 index 00000000..5b84a646 --- /dev/null +++ b/blockchain/indexers/txindex_test.go @@ -0,0 +1,564 @@ +// Copyright (c) 2020 The Decred developers +// Use of this source code is governed by an ISC +// license that can be found in the LICENSE file. + +package indexers + +import ( + "context" + "fmt" + "io/ioutil" + "os" + "sync" + "testing" + "time" + + "github.com/decred/dcrd/blockchain/v4/chaingen" + "github.com/decred/dcrd/blockchain/v4/internal/spendpruner" + "github.com/decred/dcrd/chaincfg/chainhash" + "github.com/decred/dcrd/chaincfg/v3" + "github.com/decred/dcrd/database/v2" + _ "github.com/decred/dcrd/database/v2/ffldb" + "github.com/decred/dcrd/dcrutil/v4" + "github.com/decred/dcrd/wire" +) + +// testChain represents a mock implementation of a block chain as +// defined by the indexer.ChainQueryer interface. +type testChain struct { + bestHeight int64 + bestHash *chainhash.Hash + treasuryActive bool + keyedByHeight map[int64]*dcrutil.Block + keyedByHash map[string]*dcrutil.Block + orphans map[string]*dcrutil.Block + mtx sync.Mutex +} + +// newTestChain initializes a test chain. +func newTestChain() (*testChain, error) { + tc := &testChain{ + keyedByHeight: make(map[int64]*dcrutil.Block), + keyedByHash: make(map[string]*dcrutil.Block), + orphans: make(map[string]*dcrutil.Block), + } + genesis := dcrutil.NewBlock(chaincfg.SimNetParams().GenesisBlock) + return tc, tc.AddBlock(genesis) +} + +// AddBlock extends the chain with the provided block. +func (tc *testChain) AddBlock(blk *dcrutil.Block) error { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + // Ensure the incoming block is the child of the current chain tip. + if tc.bestHash != nil { + if blk.MsgBlock().Header.PrevBlock != *tc.bestHash { + return fmt.Errorf("block %s is an orphan", blk.Hash().String()) + } + } + + height := blk.Height() + hash := blk.Hash() + tc.keyedByHash[hash.String()] = blk + tc.keyedByHeight[height] = blk + tc.bestHeight = height + tc.bestHash = hash + + return nil +} + +// Remove block disconnects the provided block as the tip +// of the chain. The provided block is required to be the +// current chain tip. +func (tc *testChain) RemoveBlock(blk *dcrutil.Block) error { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + hash := blk.Hash() + + // Ensure the block being removed is the current chain tip. + if *tc.bestHash != *hash { + return fmt.Errorf("block %s is not the current chain tip", + blk.Hash().String()) + } + + // Set the new chain tip. + tc.bestHash = &blk.MsgBlock().Header.PrevBlock + tc.bestHeight-- + + height := blk.Height() + delete(tc.keyedByHash, hash.String()) + delete(tc.keyedByHeight, height) + + tc.orphans[hash.String()] = blk + + return nil +} + +// MainChainHasBlock asserts if the provided block is part of the +// chain. +func (tc *testChain) MainChainHasBlock(hash *chainhash.Hash) bool { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + _, ok := tc.keyedByHash[hash.String()] + return ok +} + +// Best returns the height and block hash of the the current +// chain tip. +func (tc *testChain) Best() (int64, *chainhash.Hash) { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + return tc.bestHeight, tc.bestHash +} + +// IsTreasuryAgendaActive returns whether or not the treasury agenda is active. +func (tc *testChain) IsTreasuryAgendaActive(hash *chainhash.Hash) (bool, error) { + return tc.treasuryActive, nil +} + +// BlockHeightByHash returns the height of the provided block hash if it is +// part of the chain. +func (tc *testChain) BlockHeightByHash(hash *chainhash.Hash) (int64, error) { + blk, err := tc.BlockByHash(hash) + if err != nil { + return 0, err + } + + return blk.Height(), nil +} + +// Ancestor returns the ancestor block hash of the provided block at the +// provided height. +func (tc *testChain) Ancestor(block *chainhash.Hash, height int64) *chainhash.Hash { + blk, err := tc.BlockByHash(block) + if err != nil { + log.Error(err) + return nil + } + + tipHeight := blk.Height() + + // if the provided height is greater than the chain's tip height, + // the associated block cannot be an ancestor. + if height > tipHeight { + return nil + } + + for { + if blk.Height() == height { + return blk.Hash() + } + + if blk.Height() < height { + return nil + } + + prev := &blk.MsgBlock().Header.PrevBlock + blk, err = tc.BlockByHash(prev) + if err != nil { + log.Error(err) + return nil + } + } +} + +// AddSpendConsumer adds the provided spend consumer. +func (tc *testChain) AddSpendConsumer(consumer spendpruner.SpendConsumer) { + // Do nothing. +} + +// ChainParams returns the parameters of the chain. +func (tc *testChain) ChainParams() *chaincfg.Params { + return chaincfg.SimNetParams() +} + +// BlockHashByHeight returns the block hash of the block at the provided +// height. +func (tc *testChain) BlockHashByHeight(height int64) (*chainhash.Hash, error) { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + blk := tc.keyedByHeight[height] + if blk == nil { + return nil, fmt.Errorf("no block found with height %d", height) + } + + return blk.Hash(), nil +} + +// BlockByHash returns the block associated with the provided block hash. +func (tc *testChain) BlockByHash(hash *chainhash.Hash) (*dcrutil.Block, error) { + tc.mtx.Lock() + defer tc.mtx.Unlock() + + blk := tc.keyedByHash[hash.String()] + if blk == nil { + blk = tc.orphans[hash.String()] + if blk == nil { + return nil, fmt.Errorf("no block found with hash %s", hash.String()) + } + } + + return blk, nil +} + +// PrevScripts returns a source of previous transaction scripts and their +// associated versions spent by the provided block. +func (tc *testChain) PrevScripts(database.Tx, *dcrutil.Block) (PrevScripter, error) { + return nil, nil +} + +// notifyAndWait sends the provided notification and waits for done signal +// with a one second timeout. +func notifyAndWait(t *testing.T, subber *IndexSubscriber, ntfn *IndexNtfn) { + t.Helper() + done := make(chan bool) + ntfn.Done = done + subber.Notify(ntfn) + select { + case <-done: + // Nothing to do. + case <-time.After(time.Second): + t.Fatalf("timeout waiting for done signal for notification") + } +} + +// addBlock extends the provided chain with a generated block. +func addBlock(t *testing.T, chain *testChain, gen *chaingen.Generator, name string) *dcrutil.Block { + firstBlock := gen.TipName() == "genesis" + + var msgBlk *wire.MsgBlock + if firstBlock { + msgBlk = gen.CreateBlockOne(name, 0) + } + + if msgBlk == nil { + msgBlk = gen.NextBlock(name, nil, nil) + gen.SaveTipCoinbaseOuts() + } + + blk := dcrutil.NewBlock(msgBlk) + + err := chain.AddBlock(blk) + if err != nil { + t.Fatal(err) + } + + return blk +} + +// setupDB initializes the test database. +func setupDB(t *testing.T, dbName string) (database.DB, string) { + dbPath, err := ioutil.TempDir("", dbName) + if err != nil { + t.Fatalf("unable to create test db path: %v", err) + } + + db, err := database.Create("ffldb", dbPath, wire.SimNet) + if err != nil { + os.RemoveAll(dbPath) + t.Fatalf("error creating db: %v", err) + } + + return db, dbPath +} + +// teardownDB terminates and purges the test database. +func teardownDB(db database.DB, dbPath string) { + db.Close() + os.RemoveAll(dbPath) +} + +// TestTxIndexAsync ensures the tx index behaves as expected recieving +// async notifications. +func TestTxIndexAsync(t *testing.T) { + db, path := setupDB(t, "test_txindex") + defer teardownDB(db, path) + + chain, err := newTestChain() + if err != nil { + t.Fatal(err) + } + g, err := chaingen.MakeGenerator(chaincfg.SimNetParams()) + if err != nil { + t.Fatalf("unexpected error %v", err) + } + + // Add three blocks to the chain. + addBlock(t, chain, &g, "bk1") + addBlock(t, chain, &g, "bk2") + bk3 := addBlock(t, chain, &g, "bk3") + + // Initialize the tx index. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + subber := NewIndexSubscriber(ctx) + go subber.Run(ctx) + + idx, err := NewTxIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + // Ensure the index got synced to bk3 on initialization. + tipHeight, tipHash, err := idx.Tip() + if err != nil { + t.Fatal(err) + } + + if tipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk3.Height(), tipHeight) + } + + if *tipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk3.Hash().String(), tipHash.String()) + } + + // Ensure the index remains in sync with the main chain when new + // blocks are connected. + bk4 := addBlock(t, chain, &g, "bk4") + ntfn := &IndexNtfn{ + NtfnType: ConnectNtfn, + Block: bk4, + Parent: bk3, + PrevScripts: nil, + } + notifyAndWait(t, subber, ntfn) + + bk5 := addBlock(t, chain, &g, "bk5") + ntfn = &IndexNtfn{ + NtfnType: ConnectNtfn, + Block: bk5, + Parent: bk4, + PrevScripts: nil, + } + notifyAndWait(t, subber, ntfn) + + // Ensure the index got synced to bk5. + tipHeight, tipHash, err = idx.Tip() + if err != nil { + t.Fatal(err) + } + + if tipHeight != bk5.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk5.Height(), tipHeight) + } + + if *tipHash != *bk5.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk5.Hash().String(), tipHash.String()) + } + + // Simulate a reorg by setting bk4 as the main chain tip. bk5 is now + // an orphan block. + g.SetTip("bk4") + err = chain.RemoveBlock(bk5) + if err != nil { + t.Fatal(err) + } + + // Add bk5a to the main chain. + bk5a := addBlock(t, chain, &g, "bk5a") + + // Resubscribe the index. + err = idx.sub.stop() + if err != nil { + t.Fatal(err) + } + + idx, err = NewTxIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + tipHeight, tipHash, err = idx.Tip() + if err != nil { + t.Fatal(err) + } + + // Ensure the index recovered to bk4 and synced back to the main chain tip + // bk5a. + if tipHeight != bk5a.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk5a.Height(), tipHeight) + } + + if *tipHash != *bk5a.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk5a.Hash().String(), tipHash.String()) + } + + // Ensure bk5 is no longer indexed. + entry, err := idx.Entry(bk5.Hash()) + if err != nil { + t.Fatal(err) + } + + if entry != nil { + t.Fatal("expected no index entry for bk5") + } + + // Ensure the index remains in sync when blocks are disconnected. + err = chain.RemoveBlock(bk5a) + if err != nil { + t.Fatal(err) + } + + g.SetTip("bk4") + + ntfn = &IndexNtfn{ + NtfnType: DisconnectNtfn, + Block: bk5a, + Parent: bk4, + PrevScripts: nil, + } + notifyAndWait(t, subber, ntfn) + + err = chain.RemoveBlock(bk4) + if err != nil { + t.Fatal(err) + } + + g.SetTip("bk3") + + ntfn = &IndexNtfn{ + NtfnType: DisconnectNtfn, + Block: bk4, + Parent: bk3, + PrevScripts: nil, + } + notifyAndWait(t, subber, ntfn) + + // Ensure the index tip is now bk3 after the disconnections. + tipHeight, tipHash, err = idx.Tip() + if err != nil { + t.Fatal(err) + } + + if tipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk3.Height(), tipHeight) + } + + if *tipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk3.Hash().String(), tipHash.String()) + } + + // Drop the index. + err = idx.DropIndex(ctx, idx.db) + if err != nil { + t.Fatal(err) + } + + // Resubscribe the index. + err = idx.sub.stop() + if err != nil { + t.Fatal(err) + } + + idx, err = NewTxIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + // Ensure the index got synced to bk3 on initialization. + tipHeight, tipHash, err = idx.Tip() + if err != nil { + t.Fatal(err) + } + + if tipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk3.Height(), tipHeight) + } + + if *tipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk3.Hash().String(), tipHash.String()) + } + + // Add bk4a to the main chain. + bk4a := addBlock(t, chain, &g, "bk4a") + + go func() { + // Stall the index notification for bk4a. + time.Sleep(time.Millisecond * 150) + notif := &IndexNtfn{ + NtfnType: ConnectNtfn, + Block: bk4a, + Parent: bk3, + PrevScripts: nil, + Done: make(chan bool), + } + subber.Notify(notif) + select { + case <-notif.Done: + // Nothing to do. + case <-time.After(time.Second): + panic("timeout waiting for done signal for notification") + } + }() + + // Wait for the index to sync with the main chain before terminating. + select { + case <-idx.WaitForSync(): + // Nothing to do. + case <-time.After(time.Second): + panic("timeout waiting for index to synchronize") + } + + // Add bk6 and bk7 to the main chain. + bk6 := addBlock(t, chain, &g, "bk6") + bk7 := addBlock(t, chain, &g, "bk7") + + // Ensure sending an unexpected index notification (bk7) does not + // update the index. + ntfn = &IndexNtfn{ + NtfnType: ConnectNtfn, + Block: bk7, + Parent: bk6, + PrevScripts: nil, + } + notifyAndWait(t, subber, ntfn) + + // Ensure the address index remains at tip bk4a after receiving unexpected + // index notification for bk7. + tipHeight, tipHash, err = idx.Tip() + if err != nil { + t.Fatal(err) + } + + if tipHeight != bk4a.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk4a.Height(), tipHeight) + } + + if *tipHash != *bk4a.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk4a.Hash().String(), tipHash.String()) + } +}