From 35a230790cc72cf67937bd98162f01feaa5ee3df Mon Sep 17 00:00:00 2001 From: Donald Adu-Poku Date: Sun, 13 Jun 2021 19:06:47 +0000 Subject: [PATCH] indexers: update address index. This updates the AddressIndex to implement the updated Indexer interface for index updates. The index however subscribes with the transaction index as a prerequisite. This also adds the spend consumer for use by indexers that need to lookup spend journal entries. --- blockchain/indexers/addrindex.go | 183 +++++++-- blockchain/indexers/addrindex_test.go | 430 ++++++++++++++++++++++ blockchain/indexers/spendconsumer.go | 80 ++++ blockchain/indexers/spendconsumer_test.go | 100 +++++ 4 files changed, 765 insertions(+), 28 deletions(-) create mode 100644 blockchain/indexers/spendconsumer.go create mode 100644 blockchain/indexers/spendconsumer_test.go diff --git a/blockchain/indexers/addrindex.go b/blockchain/indexers/addrindex.go index 7268b0ec..8edc772f 100644 --- a/blockchain/indexers/addrindex.go +++ b/blockchain/indexers/addrindex.go @@ -582,7 +582,10 @@ type AddrIndex struct { // be changed afterwards, so there is no need to protect them with a // separate mutex. db database.DB + chain ChainQueryer chainParams *chaincfg.Params + sub *IndexSubscription + consumer *SpendConsumer // The following fields are used to quickly link transactions and // addresses that have not been included into a block yet when an @@ -600,6 +603,10 @@ type AddrIndex struct { unconfirmedLock sync.RWMutex txnsByAddr map[[addrKeySize]byte]map[chainhash.Hash]*dcrutil.Tx addrsByTx map[chainhash.Hash]map[[addrKeySize]byte]struct{} + + subscribers map[chan bool]struct{} + mtx sync.Mutex + cancel context.CancelFunc } // Ensure the AddrIndex type implements the Indexer interface. @@ -616,12 +623,36 @@ func (idx *AddrIndex) NeedsInputs() bool { return true } -// Init is only provided to satisfy the Indexer interface as there is nothing to -// initialize for this index. +// Init creates a transaction by address index. In particular, it maintains +// a map of transactions and their associated addresses via a stream of updates +// on connected and disconnected blocks. // // This is part of the Indexer interface. -func (idx *AddrIndex) Init() error { - // Nothing to do. +func (idx *AddrIndex) 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 address index and its dependents to the main chain if needed. + if err := recover(ctx, idx); err != nil { + return err + } + return nil } @@ -646,9 +677,58 @@ func (idx *AddrIndex) Version() uint32 { return addrIndexVersion } -// Create is invoked when the indexer manager determines the index needs -// to be created for the first time. It creates the bucket for the address -// index. +// DB returns the database of the index. +// +// This is part of the Indexer interface. +func (idx *AddrIndex) DB() database.DB { + return idx.db +} + +// Queryer returns the chain queryer. +// +// This is part of the Indexer interface. +func (idx *AddrIndex) Queryer() ChainQueryer { + return idx.chain +} + +// Tip returns the current tip of the index. +// +// This is part of the Indexer interface. +func (idx *AddrIndex) 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 *AddrIndex) 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 *AddrIndex) 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 *AddrIndex) 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 bucket for the address index. // // This is part of the Indexer interface. func (idx *AddrIndex) Create(dbTx database.Tx) error { @@ -791,12 +871,9 @@ func (idx *AddrIndex) indexBlock(data writeIndexData, block *dcrutil.Block, prev } } -// ConnectBlock is invoked by the index manager when a new block has been -// connected to the main chain. This indexer adds a mapping for each address -// the transactions in the block involve. -// -// This is part of the Indexer interface. -func (idx *AddrIndex) ConnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, prevScripts PrevScripter, isTreasuryEnabled bool) error { +// connectBlock adds a mapping for all addresses associated with transactions in +// the provided block. +func (idx *AddrIndex) connectBlock(dbTx database.Tx, block, parent *dcrutil.Block, prevScripts PrevScripter, isTreasuryEnabled 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 @@ -841,15 +918,13 @@ func (idx *AddrIndex) ConnectBlock(dbTx database.Tx, block, parent *dcrutil.Bloc } } - 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 address mappings -// each transaction in the block involve. -// -// This is part of the Indexer interface. -func (idx *AddrIndex) DisconnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, prevScripts PrevScripter, isTreasuryEnabled bool) error { +// disconnectBlock removes the mappings for addresses associated with +// transactions in the provided block. +func (idx *AddrIndex) disconnectBlock(dbTx database.Tx, block, parent *dcrutil.Block, prevScripts PrevScripter, isTreasuryEnabled 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 @@ -869,7 +944,9 @@ func (idx *AddrIndex) DisconnectBlock(dbTx database.Tx, block, parent *dcrutil.B } } - return nil + // Update the current index tip. + return dbPutIndexerTip(dbTx, idx.Key(), &block.MsgBlock().Header.PrevBlock, + int32(block.Height()-1)) } // EntriesForAddress returns a slice of details which identify each transaction, @@ -1055,17 +1132,35 @@ func (idx *AddrIndex) UnconfirmedTxnsForAddress(addr stdaddr.Address) []*dcrutil // NewAddrIndex returns a new instance of an indexer that is used to create a // mapping of all addresses in the blockchain to the respective transactions // that involve them. -// -// 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 NewAddrIndex(db database.DB, chainParams *chaincfg.Params) *AddrIndex { - return &AddrIndex{ +func NewAddrIndex(subscriber *IndexSubscriber, db database.DB, chain ChainQueryer) (*AddrIndex, error) { + idx := &AddrIndex{ db: db, - chainParams: chainParams, + chain: chain, + chainParams: chain.ChainParams(), + subscribers: make(map[chan bool]struct{}), txnsByAddr: make(map[[addrKeySize]byte]map[chainhash.Hash]*dcrutil.Tx), addrsByTx: make(map[chainhash.Hash]map[[addrKeySize]byte]struct{}), + cancel: subscriber.cancel, } + + idx.consumer = NewSpendConsumer(idx.Name(), nil, chain) + chain.AddSpendConsumer(idx.consumer) + + // The address index is an optional index. It depends on the + // transaction index and as a result synchronously updates with it. + sub, err := subscriber.Subscribe(idx, txIndexName) + if err != nil { + return nil, err + } + + idx.sub = sub + + err = idx.Init(subscriber.ctx, idx.chain.ChainParams()) + if err != nil { + return nil, err + } + + return idx, nil } // DropAddrIndex drops the address index from the provided database if it @@ -1078,3 +1173,35 @@ func DropAddrIndex(ctx context.Context, db database.DB) error { func (*AddrIndex) DropIndex(ctx context.Context, db database.DB) error { return DropAddrIndex(ctx, db) } + +// ProcessNotification indexes the provided notification based on its +// notification type. +// +// This is part of the Indexer interface. +func (idx *AddrIndex) 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) + } + + idx.consumer.UpdateTip(ntfn.Block.Hash()) + + 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) + } + + idx.consumer.UpdateTip(ntfn.Parent.Hash()) + + default: + return fmt.Errorf("%s: unknown notification type provided: %d", + idx.Name(), ntfn.NtfnType) + } + + return nil +} diff --git a/blockchain/indexers/addrindex_test.go b/blockchain/indexers/addrindex_test.go index 7a37dd8d..bae00a05 100644 --- a/blockchain/indexers/addrindex_test.go +++ b/blockchain/indexers/addrindex_test.go @@ -7,9 +7,15 @@ package indexers import ( "bytes" + "context" "fmt" "testing" + "time" + "github.com/decred/dcrd/blockchain/v4/chaingen" + "github.com/decred/dcrd/chaincfg/v3" + "github.com/decred/dcrd/database/v2" + "github.com/decred/dcrd/txscript/v4" "github.com/decred/dcrd/wire" ) @@ -274,3 +280,427 @@ nextTest: } } } + +// TestAddrIndexAsync ensures the address index behaves +// receiving updates asynchronously. +func TestAddrIndexAsync(t *testing.T) { + db, path := setupDB(t, "test_addrindex") + defer teardownDB(db, path) + + // Create the test chain. + 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 and address index. The tx index is required + // because the address index depends on it. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + subber := NewIndexSubscriber(ctx) + go subber.Run(ctx) + + txIdx, err := NewTxIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + addrIdx, err := NewAddrIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + // Ensure the indexes got synced to bk3 on initialization. + txIdxTipHeight, txIdxTipHash, err := addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if txIdxTipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk3.Height(), txIdxTipHeight) + } + + if *txIdxTipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk3.Hash().String(), + txIdxTipHash.String()) + } + + addrIdxTipHeight, addrIdxTipHash, err := addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk3.Height(), addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk3.Hash().String(), + addrIdxTipHash.String()) + } + + // Ensure the address 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, + IsTreasuryEnabled: false, + } + notifyAndWait(t, subber, ntfn) + + bk5 := addBlock(t, chain, &g, "bk5") + ntfn = &IndexNtfn{ + NtfnType: ConnectNtfn, + Block: bk5, + Parent: bk4, + PrevScripts: nil, + IsTreasuryEnabled: false, + } + notifyAndWait(t, subber, ntfn) + + // Ensure the indexes got synced to bk5. + txIdxTipHeight, txIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if txIdxTipHeight != bk5.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk5.Height(), + txIdxTipHeight) + } + + if *txIdxTipHash != *bk5.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk5.Hash().String(), + txIdxTipHash.String()) + } + + addrIdxTipHeight, addrIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk5.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk5.Height(), + addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk5.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk5.Hash().String(), + addrIdxTipHash.String()) + } + + // Ensure the spend consumer's tip was updated to bk5. + addrIdx.consumer.mtx.Lock() + tipHash := addrIdx.consumer.tipHash + addrIdx.consumer.mtx.Unlock() + + if !tipHash.IsEqual(bk5.Hash()) { + t.Fatalf("expected spend consumer tip hash to be %s, got %s", + bk5.Hash().String(), tipHash.String()) + } + + // Fetch the first address paid to by bk5's coinbase. + out := bk5.MsgBlock().Transactions[0].TxOut[0] + _, addrs, _, err := txscript.ExtractPkScriptAddrs(out.Version, + out.PkScript, addrIdx.chainParams, true) + if err != nil { + t.Fatal(err) + } + + // Ensure the address index has the first address paid to by bk5's + // coinbase indexed. + err = db.View(func(dbTx database.Tx) error { + entry, _, err := addrIdx.EntriesForAddress(dbTx, addrs[0], 0, 1, false) + if err != nil { + return err + } + if entry == nil { + return fmt.Errorf("no index entry found for address %s", + addrs[0].String()) + } + + return nil + }) + if err != nil { + t.Fatal(err) + } + + // 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 indexes. + err = addrIdx.sub.stop() + if err != nil { + t.Fatal(err) + } + + err = txIdx.sub.stop() + if err != nil { + t.Fatal(err) + } + + txIdx, err = NewTxIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + addrIdx, err = NewAddrIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + // Ensure the indexes recovered to bk4 and synced back to the main chain tip + // bk5a. + txIdxTipHeight, txIdxTipHash, err = txIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if txIdxTipHeight != bk5a.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk5a.Height(), txIdxTipHeight) + } + + if *txIdxTipHash != *bk5a.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk5a.Hash().String(), txIdxTipHash.String()) + } + + addrIdxTipHeight, addrIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk5a.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk5a.Height(), addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk5a.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk5a.Hash().String(), addrIdxTipHash.String()) + } + + // Ensure the spend consumer's tip was updated to bk5a. + addrIdx.consumer.mtx.Lock() + tipHash = addrIdx.consumer.tipHash + addrIdx.consumer.mtx.Unlock() + + if !tipHash.IsEqual(bk5a.Hash()) { + t.Fatalf("expected spend consumer tip hash to be %s, got %s", + bk5a.Hash().String(), tipHash.String()) + } + + // Ensure the addreses associated with bk5's coinbase transaction + // is no longer indexed by the address manager. + // Ensure the address index no longder has the first address paid to by + // bk5's coinbase indexed. + err = db.View(func(dbTx database.Tx) error { + entry, _, err := addrIdx.EntriesForAddress(dbTx, addrs[0], 0, 1, false) + if err != nil { + return err + } + if entry != nil { + return fmt.Errorf("expected no index entry found for address %s", + addrs[0].String()) + } + + return nil + }) + if err == nil { + t.Fatalf("expected no index entry found for address %s", + addrs[0].String()) + } + + // Ensure the indexes remain 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, + IsTreasuryEnabled: false, + } + 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, + IsTreasuryEnabled: false, + } + notifyAndWait(t, subber, ntfn) + + // Ensure the index tips are now bk3 after the disconnections. + txIdxTipHeight, txIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if txIdxTipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk3.Height(), txIdxTipHeight) + } + + if *txIdxTipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk3.Hash().String(), txIdxTipHash.String()) + } + + addrIdxTipHeight, addrIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk3.Height(), addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk3.Hash().String(), + addrIdxTipHash.String()) + } + + // Drop the address index and resubscribe. + err = addrIdx.sub.stop() + if err != nil { + t.Fatal(err) + } + + err = addrIdx.DropIndex(ctx, db) + if err != nil { + t.Fatal(err) + } + + addrIdx, err = NewAddrIndex(subber, db, chain) + if err != nil { + t.Fatal(err) + } + + err = subber.CatchUp(ctx, db, chain) + if err != nil { + t.Fatal(err) + } + + // Ensure the address index got synced to bk3 on initialization. + addrIdxTipHeight, addrIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk3.Height() { + t.Fatalf("expected tip height to be %d, got %d", bk3.Height(), addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk3.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", bk3.Hash().String(), + addrIdxTipHash.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. + select { + case <-addrIdx.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. + addrIdxTipHeight, addrIdxTipHash, err = addrIdx.Tip() + if err != nil { + t.Fatal(err) + } + + if addrIdxTipHeight != bk4a.Height() { + t.Fatalf("expected tip height to be %d, got %d", + bk4a.Height(), addrIdxTipHeight) + } + + if *addrIdxTipHash != *bk4a.Hash() { + t.Fatalf("expected tip hash to be %s, got %s", + bk4a.Hash().String(), addrIdxTipHash.String()) + } +} diff --git a/blockchain/indexers/spendconsumer.go b/blockchain/indexers/spendconsumer.go new file mode 100644 index 00000000..cff0095a --- /dev/null +++ b/blockchain/indexers/spendconsumer.go @@ -0,0 +1,80 @@ +// Copyright (c) 2021 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 ( + "sync" + + "github.com/decred/dcrd/chaincfg/chainhash" +) + +// SpendConsumer represents an indexer dependent on +// spend journal entries. +type SpendConsumer struct { + id string + queryer ChainQueryer + tipHash *chainhash.Hash + mtx sync.Mutex +} + +// NewSpendConsumer initializes a spend consumer. +func NewSpendConsumer(id string, tipHash *chainhash.Hash, queryer ChainQueryer) *SpendConsumer { + return &SpendConsumer{ + id: id, + queryer: queryer, + tipHash: tipHash, + } +} + +// ID returns the identifier of the consumer. +func (s *SpendConsumer) ID() string { + return s.id +} + +// UpdateTip sets the tip of the consumer to the provided hash. +func (s *SpendConsumer) UpdateTip(hash *chainhash.Hash) { + s.mtx.Lock() + s.tipHash = hash + s.mtx.Unlock() +} + +// NeedSpendData checks whether the associated spend journal entry +// for the provided block hash will be needed by the indexer. +func (s *SpendConsumer) NeedSpendData(hash *chainhash.Hash) (bool, error) { + // The indexer does not need spend journal data if it has not + // been initialized. + s.mtx.Lock() + tipHash := s.tipHash + s.mtx.Unlock() + if tipHash == nil { + return false, nil + } + + tipHeight, err := s.queryer.BlockHeightByHash(tipHash) + if err != nil { + return false, err + } + height, err := s.queryer.BlockHeightByHash(hash) + if err != nil { + return false, err + } + + // The spend consumer does not need the spend journal + // data associated with the provided block hash if + // its current tip is below the provided hash. + if height > tipHeight { + return false, nil + } + + // The spend consumer does not need the spend data associated + // with the provided block hash if the block is not an ancestor + // of the current tip. + blockHash := s.queryer.Ancestor(tipHash, height) + if !blockHash.IsEqual(hash) { + return false, nil + } + + return true, nil +} diff --git a/blockchain/indexers/spendconsumer_test.go b/blockchain/indexers/spendconsumer_test.go new file mode 100644 index 00000000..be32ad04 --- /dev/null +++ b/blockchain/indexers/spendconsumer_test.go @@ -0,0 +1,100 @@ +// Copyright (c) 2021 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 ( + "io/ioutil" + "os" + "testing" + + "github.com/decred/dcrd/blockchain/v4/chaingen" + "github.com/decred/dcrd/chaincfg/chainhash" + "github.com/decred/dcrd/chaincfg/v3" + "github.com/decred/dcrd/database/v2" + "github.com/decred/dcrd/wire" +) + +// TestSpendConsumer ensures the spend consumer behaves as expected. +func TestSpendConsumer(t *testing.T) { + dbPath, err := ioutil.TempDir("", "test_spendconsumer") + 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) + } + defer func() { + db.Close() + os.RemoveAll(dbPath) + }() + + 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 four blocks to the chain. + bk1 := addBlock(t, chain, &g, "bk1") + bk2 := addBlock(t, chain, &g, "bk2") + bk3 := addBlock(t, chain, &g, "bk3") + bk4 := addBlock(t, chain, &g, "bk4") + + id := "testConsumer" + consumer := NewSpendConsumer(id, bk3.Hash(), chain) + + // Ensure the consumer is correctly identified. + if id != consumer.ID() { + t.Fatalf("expected consumer id to be %v, got %v", id, consumer.ID()) + } + + // Ensure bk2 is needed by the consumer since it is an ancestor + // of the current tip, bk3. + needed, err := consumer.NeedSpendData(bk2.Hash()) + if err != nil { + t.Fatalf("unexpected NeedSpendData error: %v", err) + } + + if !needed { + t.Fatalf("expected the spend data for block bk2 to be needed") + } + + // Ensure bk4 is not needed by the address index since it is an ancestor + // of the current tip, bk4. + needed, err = consumer.NeedSpendData(bk4.Hash()) + if err != nil { + t.Fatalf("unexpected NeedSpendData error: %v", err) + } + + if needed { + t.Fatalf("expected the spend data for block bk4 to not be needed") + } + + // Ensure finding the ancestor of an unknown block hash fails. + _, err = consumer.NeedSpendData(&chainhash.Hash{0}) + if err == nil { + t.Fatal("expected an unknown block error") + } + + // Update the consumer tip to bk1. + consumer.UpdateTip(bk1.Hash()) + + // Ensure bk2 is now no longer needed by the consumer since it is + // not an ancestor of the current tip, bk1. + needed, err = consumer.NeedSpendData(bk2.Hash()) + if err != nil { + t.Fatalf("unexpected NeedSpendData error: %v", err) + } + + if needed { + t.Fatalf("expected the spend data for block bk2 to not be needed") + } +}