seedhammer/dcrd
seedhammer/dcrd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

indexers: update address index.

Browse Source 
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.
This commit is contained in:
Donald Adu-Poku 2021-06-13 19:06:47 +00:00 committed by Dave Collins
parent eeaa6d71dc
commit 35a230790c
4 changed files with 765 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

183
blockchain/indexers/addrindex.go
View File

@ -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
}

430
blockchain/indexers/addrindex_test.go
View File

@ -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())
}
}

80
blockchain/indexers/spendconsumer.go Normal file
View File

@ -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
}

100
blockchain/indexers/spendconsumer_test.go Normal file
View File

@ -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")
}
}