indexers: async transaction index.
This reworks the TransactionIndex to implement the updated Indexer interface for asynchoronous index updates. This index subscribes with no prequisites.
This commit is contained in:
Donald Adu-Poku
Dave Collins
parent
3e1090e0cc
commit
eeaa6d71dc
@ -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
|
||||
}
|
||||
|
||||
564
blockchain/indexers/txindex_test.go
Normal file
564
blockchain/indexers/txindex_test.go
Normal file
@ -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())
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user