ethstats, les, light: les txpool rewrite

core/tx_journal.go

@@ -40,23 +40,23 @@ type devNull struct{}
 func (*devNull) Write(p []byte) (n int, err error) { return len(p), nil }
 func (*devNull) Close() error                      { return nil }
 
-// txJournal is a rotating log of transactions with the aim of storing locally
+// TxJournal is a rotating log of transactions with the aim of storing locally
 // created transactions to allow non-executed ones to survive node restarts.
-type txJournal struct {
+type TxJournal struct {
 	path   string         // Filesystem path to store the transactions at
 	writer io.WriteCloser // Output stream to write new transactions into
 }
 
-// newTxJournal creates a new transaction journal to
+// NewTxJournal creates a new transaction journal to
-func newTxJournal(path string) *txJournal {
+func NewTxJournal(path string) *TxJournal {
-	return &txJournal{
+	return &TxJournal{
 		path: path,
 	}
 }
 
-// load parses a transaction journal dump from disk, loading its contents into
+// Load parses a transaction journal dump from disk, loading its contents into
 // the specified pool.
-func (journal *txJournal) load(add func([]*types.Transaction) []error) error {
+func (journal *TxJournal) Load(add func([]*types.Transaction) []error) error {
 	// Skip the parsing if the journal file doesn't exist at all
 	if _, err := os.Stat(journal.path); os.IsNotExist(err) {
 		return nil
@@ -116,8 +116,8 @@ func (journal *txJournal) load(add func([]*types.Transaction) []error) error {
 	return failure
 }
 
-// insert adds the specified transaction to the local disk journal.
+// Insert adds the specified transaction to the local disk journal.
-func (journal *txJournal) insert(tx *types.Transaction) error {
+func (journal *TxJournal) Insert(tx *types.Transaction) error {
 	if journal.writer == nil {
 		return errNoActiveJournal
 	}
@@ -127,9 +127,9 @@ func (journal *txJournal) insert(tx *types.Transaction) error {
 	return nil
 }
 
-// rotate regenerates the transaction journal based on the current contents of
+// Rotate regenerates the transaction journal based on the current contents of
 // the transaction pool.
-func (journal *txJournal) rotate(all map[common.Address]types.Transactions) error {
+func (journal *TxJournal) Rotate(all map[common.Address]types.Transactions) error {
 	// Close the current journal (if any is open)
 	if journal.writer != nil {
 		if err := journal.writer.Close(); err != nil {
@@ -168,8 +168,8 @@ func (journal *txJournal) rotate(all map[common.Address]types.Transactions) erro
 	return nil
 }
 
-// close flushes the transaction journal contents to disk and closes the file.
+// Close flushes the transaction journal contents to disk and closes the file.
-func (journal *txJournal) close() error {
+func (journal *TxJournal) Close() error {
 	var err error
 
 	if journal.writer != nil {

core/tx_pool.go

@@ -219,7 +219,7 @@ type TxPool struct {
 	currentMaxGas uint64              // Current gas limit for transaction caps
 
 	locals  *accountSet // Set of local transaction to exempt from eviction rules
-	journal *txJournal  // Journal of local transaction to back up to disk
+	journal *TxJournal  // Journal of local transaction to back up to disk
 
 	pending map[common.Address]*txList   // All currently processable transactions
 	queue   map[common.Address]*txList   // Queued but non-processable transactions
@@ -261,12 +261,12 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
 
 	// If local transactions and journaling is enabled, load from disk
 	if !config.NoLocals && config.Journal != "" {
-		pool.journal = newTxJournal(config.Journal)
+		pool.journal = NewTxJournal(config.Journal)
 
-		if err := pool.journal.load(pool.AddLocals); err != nil {
+		if err := pool.journal.Load(pool.AddLocals); err != nil {
 			log.Warn("Failed to load transaction journal", "err", err)
 		}
-		if err := pool.journal.rotate(pool.local()); err != nil {
+		if err := pool.journal.Rotate(pool.local()); err != nil {
 			log.Warn("Failed to rotate transaction journal", "err", err)
 		}
 	}
@@ -353,7 +353,7 @@ func (pool *TxPool) loop() {
 		case <-journal.C:
 			if pool.journal != nil {
 				pool.mu.Lock()
-				if err := pool.journal.rotate(pool.local()); err != nil {
+				if err := pool.journal.Rotate(pool.local()); err != nil {
 					log.Warn("Failed to rotate local tx journal", "err", err)
 				}
 				pool.mu.Unlock()
@@ -480,7 +480,7 @@ func (pool *TxPool) Stop() {
 	pool.wg.Wait()
 
 	if pool.journal != nil {
-		pool.journal.close()
+		pool.journal.Close()
 	}
 	log.Info("Transaction pool stopped")
 }
@@ -759,7 +759,7 @@ func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) {
 	if pool.journal == nil || !pool.locals.contains(from) {
 		return
 	}
-	if err := pool.journal.insert(tx); err != nil {
+	if err := pool.journal.Insert(tx); err != nil {
 		log.Warn("Failed to journal local transaction", "err", err)
 	}
 }

ethstats/ethstats.go

@@ -644,7 +644,7 @@ func (s *Service) reportPending(conn *websocket.Conn) error {
 	if s.eth != nil {
 		pending, _ = s.eth.TxPool().Stats()
 	} else {
-		pending = s.les.TxPool().Stats()
+		pending, _ = s.les.TxPool().GetPending()
 	}
 	// Assemble the transaction stats and send it to the server
 	log.Trace("Sending pending transactions to ethstats", "count", pending)

les/api_backend.go

@@ -120,16 +120,13 @@ func (b *LesApiBackend) SendTx(ctx context.Context, signedTx *types.Transaction)
 	return b.eth.txPool.Add(ctx, signedTx)
 }
 
-func (b *LesApiBackend) RemoveTx(txHash common.Hash) {
-	b.eth.txPool.RemoveTx(txHash)
-}
-
 func (b *LesApiBackend) GetPoolTransactions() (types.Transactions, error) {
-	return b.eth.txPool.GetTransactions()
+	return b.eth.txPool.GetAllPendingTransactions()
 }
 
 func (b *LesApiBackend) GetPoolTransaction(txHash common.Hash) *types.Transaction {
-	return b.eth.txPool.GetTransaction(txHash)
+	tx, _ := b.eth.txPool.GetPendingTransaction(txHash)
+	return tx
 }
 
 func (b *LesApiBackend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
@@ -141,7 +138,8 @@ func (b *LesApiBackend) GetPoolNonce(ctx context.Context, addr common.Address) (
 }
 
 func (b *LesApiBackend) Stats() (pending int, queued int) {
-	return b.eth.txPool.Stats(), 0
+	pending, _ = b.eth.txPool.GetPending()
+	return pending, 0
 }
 
 func (b *LesApiBackend) TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {

les/backend.go

@@ -140,7 +140,10 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 		rawdb.WriteChainConfig(chainDb, genesisHash, chainConfig)
 	}
 
-	leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay)
+	if config.TxPool.Journal != "" {
+		config.TxPool.Journal = ctx.ResolvePath(config.TxPool.Journal)
+	}
+	leth.txPool = light.NewTxPool(config.TxPool, leth.chainConfig, leth.blockchain, leth.relay)
 
 	if leth.protocolManager, err = NewProtocolManager(
 		leth.chainConfig,

les/peer.go

@@ -464,7 +464,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 	send = send.add("genesisHash", genesis)
 	if server != nil {
 		if !server.onlyAnnounce {
-			//only announce server. It sends only announse requests
+			// only announce server. It sends only announse requests
 			send = send.add("serveHeaders", nil)
 			send = send.add("serveChainSince", uint64(0))
 			send = send.add("serveStateSince", uint64(0))
@@ -482,7 +482,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 		p.fcCosts = costList.decode(ProtocolLengths[uint(p.version)])
 		p.fcParams = server.defParams
 	} else {
-		//on client node
+		// on client node
 		p.announceType = announceTypeSimple
 		if p.isTrusted {
 			p.announceType = announceTypeSigned
@@ -538,22 +538,18 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 			return errResp(ErrUselessPeer, "wanted client, got server")
 		}*/
 		if recv.get("announceType", &p.announceType) != nil {
-			//set default announceType on server side
+			// set default announceType on server side
 			p.announceType = announceTypeSimple
 		}
 		p.fcClient = flowcontrol.NewClientNode(server.fcManager, server.defParams)
 	} else {
-		//mark OnlyAnnounce server if "serveHeaders", "serveChainSince", "serveStateSince" or "txRelay" fields don't exist
+		// mark OnlyAnnounce server if "serveHeaders", "serveChainSince", "serveStateSince" or "txRelay" fields don't exist
-		if recv.get("serveChainSince", nil) != nil {
+		serviceList := []string{"serveHeaders", "serveChainSince", "serveStateSince", "txRelay"}
+		for i := 0; i < len(serviceList); i++ {
+			if recv.get(serviceList[i], nil) != nil {
 				p.isOnlyAnnounce = true
 			}
-		if recv.get("serveStateSince", nil) != nil {
-			p.isOnlyAnnounce = true
 		}
-		if recv.get("txRelay", nil) != nil {
-			p.isOnlyAnnounce = true
-		}
-
 		if p.isOnlyAnnounce && !p.isTrusted {
 			return errResp(ErrUselessPeer, "peer cannot serve requests")
 		}

les/txrelay.go

@@ -18,112 +18,146 @@ package les
 
 import (
 	"context"
+	"math"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/prque"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/light"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
-type ltrInfo struct {
+// relayTracer includes the raw transaction and relative
+// relay information.
+type relayTracer struct {
 	tx    *types.Transaction
-	sentTo map[*peer]struct{}
+	queue *prque.Prque  // Priority queue of the peers to relay the transactions to.
+	index map[*peer]int // Peer indexes in the priority queue used to remove element.
+}
+
+// newRelayTracer creates a relayTracer and initializes the priority queue.
+func newRelayTracer(tx *types.Transaction, peers []*peer) *relayTracer {
+	info := &relayTracer{
+		tx:    tx,
+		index: make(map[*peer]int),
+	}
+	info.queue = prque.New(info.setIndex)
+	for _, peer := range peers {
+		info.queue.Push(peer, 0)
+	}
+	return info
+}
+
+// setIndex saves the index in queue of element into the map.
+func (r *relayTracer) setIndex(a interface{}, i int) {
+	r.index[a.(*peer)] = i
 }
 
 type LesTxRelay struct {
-	txSent       map[common.Hash]*ltrInfo
-	txPending    map[common.Hash]struct{}
-	ps           *peerSet
 	peerList  []*peer
-	peerStartPos int
-	lock         sync.RWMutex
-	stop         chan struct{}
-
 	retriever *retrieveManager
+	pending   map[common.Hash]*relayTracer // Transactions which has been sent but not finalized.
+	stop      chan struct{}
+	lock      sync.RWMutex
 }
 
 func NewLesTxRelay(ps *peerSet, retriever *retrieveManager) *LesTxRelay {
 	r := &LesTxRelay{
-		txSent:    make(map[common.Hash]*ltrInfo),
+		pending:   make(map[common.Hash]*relayTracer),
-		txPending: make(map[common.Hash]struct{}),
-		ps:        ps,
 		retriever: retriever,
 		stop:      make(chan struct{}),
 	}
+	for _, peer := range ps.AllPeers() {
+		if !peer.isOnlyAnnounce {
+			r.peerList = append(r.peerList, peer)
+		}
+	}
 	ps.notify(r)
 	return r
 }
 
-func (self *LesTxRelay) Stop() {
+func (l *LesTxRelay) Stop() {
-	close(self.stop)
+	close(l.stop)
 }
 
-func (self *LesTxRelay) registerPeer(p *peer) {
+func (l *LesTxRelay) registerPeer(p *peer) {
-	self.lock.Lock()
+	l.lock.Lock()
-	defer self.lock.Unlock()
+	defer l.lock.Unlock()
 
-	self.peerList = self.ps.AllPeers()
+	// Short circuit if the peer is announce only.
-}
+	if p.isOnlyAnnounce {
-
+		return
-func (self *LesTxRelay) unregisterPeer(p *peer) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-
-	self.peerList = self.ps.AllPeers()
-}
-
-// send sends a list of transactions to at most a given number of peers at
-// once, never resending any particular transaction to the same peer twice
-func (self *LesTxRelay) send(txs types.Transactions, count int) {
-	sendTo := make(map[*peer]types.Transactions)
-
-	self.peerStartPos++ // rotate the starting position of the peer list
-	if self.peerStartPos >= len(self.peerList) {
-		self.peerStartPos = 0
 	}
+	l.peerList = append(l.peerList, p)
 
+	// Register new peer to all relay tracers.
+	for _, tx := range l.pending {
+		tx.queue.Push(p, 0)
+	}
+}
+
+func (l *LesTxRelay) unregisterPeer(p *peer) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	for i, peer := range l.peerList {
+		if peer == p {
+			// Remove from the peer list
+			l.peerList = append(l.peerList[:i], l.peerList[i+1:]...)
+
+			// Update all relay tracers as well.
+			for _, tx := range l.pending {
+				if _, exist := tx.index[p]; exist {
+					tx.queue.Remove(tx.index[p])
+					delete(tx.index, p)
+				}
+			}
+		}
+	}
+}
+
+// send relays a list of transactions to at most a given number of peers at
+// once, never resending any particular transaction to the same peer twice.
+func (l *LesTxRelay) send(txs types.Transactions) {
+	var (
+		resend = int(math.Sqrt(float64(len(l.peerList))))
+		sendTo = make(map[*peer]types.Transactions)
+	)
 	for _, tx := range txs {
 		hash := tx.Hash()
-		ltr, ok := self.txSent[hash]
+		t, exist := l.pending[hash]
+		if !exist {
+			t = newRelayTracer(tx, l.peerList)
+			l.pending[hash] = t
+		}
+		// If this is a new transaction, broadcast to all sendable peers.
+		// Otherwise(e.g. resend reverted transaction), only send to a part
+		// of them.
+		cnt := len(l.peerList)
+		if exist {
+			cnt = resend
+		}
+		for i := 0; i < cnt; i++ {
+			item, priority := t.queue.Pop()
+			peer, ok := item.(*peer)
 			if !ok {
-			ltr = &ltrInfo{
+				log.Warn("Unexpected item in priority queue")
-				tx:     tx,
+				continue
-				sentTo: make(map[*peer]struct{}),
 			}
-			self.txSent[hash] = ltr
-			self.txPending[hash] = struct{}{}
-		}
-
-		if len(self.peerList) > 0 {
-			cnt := count
-			pos := self.peerStartPos
-			for {
-				peer := self.peerList[pos]
-				if _, ok := ltr.sentTo[peer]; !ok {
 			sendTo[peer] = append(sendTo[peer], tx)
-					ltr.sentTo[peer] = struct{}{}
+			t.queue.Push(item, priority-1)
-					cnt--
-				}
-				if cnt == 0 {
-					break // sent it to the desired number of peers
-				}
-				pos++
-				if pos == len(self.peerList) {
-					pos = 0
-				}
-				if pos == self.peerStartPos {
-					break // tried all available peers
-				}
-			}
 		}
 	}
 
-	for p, list := range sendTo {
+	for p, txs := range sendTo {
-		pp := p
+		var (
-		ll := list
+			pp     = p
-		enc, _ := rlp.EncodeToBytes(ll)
+			ll     = txs
-
+			enc, _ = rlp.EncodeToBytes(txs)
-		reqID := genReqID()
+			reqID  = genReqID()
+		)
 		rq := &distReq{
 			getCost: func(dp distPeer) uint64 {
 				peer := dp.(*peer)
@@ -139,46 +173,29 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
 				return func() { peer.SendTxs(reqID, cost, enc) }
 			},
 		}
-		go self.retriever.retrieve(context.Background(), reqID, rq, func(p distPeer, msg *Msg) error { return nil }, self.stop)
+		go l.retriever.retrieve(context.Background(), reqID, rq, func(p distPeer, msg *Msg) error { return nil }, l.stop)
 	}
 }
 
-func (self *LesTxRelay) Send(txs types.Transactions) {
+// Send relays a batch of transaction into the network and returns all unsend
-	self.lock.Lock()
+// transactions.
-	defer self.lock.Unlock()
+func (l *LesTxRelay) Send(txs types.Transactions) error {
+	l.lock.Lock()
+	defer l.lock.Unlock()
 
-	self.send(txs, 3)
+	if len(l.peerList) == 0 {
+		return light.ErrNoPeers
+	}
+	l.send(txs)
+	return nil
 }
 
-func (self *LesTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
+// Discard marks a batch of transaction are finalized and won't be reverted.
-	self.lock.Lock()
+func (l *LesTxRelay) Discard(hashes []common.Hash) {
-	defer self.lock.Unlock()
+	l.lock.Lock()
-
+	defer l.lock.Unlock()
-	for _, hash := range mined {
-		delete(self.txPending, hash)
-	}
-
-	for _, hash := range rollback {
-		self.txPending[hash] = struct{}{}
-	}
-
-	if len(self.txPending) > 0 {
-		txs := make(types.Transactions, len(self.txPending))
-		i := 0
-		for hash := range self.txPending {
-			txs[i] = self.txSent[hash].tx
-			i++
-		}
-		self.send(txs, 1)
-	}
-}
-
-func (self *LesTxRelay) Discard(hashes []common.Hash) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
 
 	for _, hash := range hashes {
-		delete(self.txSent, hash)
+		delete(l.pending, hash)
-		delete(self.txPending, hash)
 	}
 }

light/odr_test.go

@@ -58,6 +58,13 @@ type testOdr struct {
 	indexerConfig *IndexerConfig
 	sdb, ldb      ethdb.Database
 	disable       bool
+
+	// Test hooks
+	isBlockHookTarget func(common.Hash) bool
+	blockHook         func(common.Hash) []byte
+
+	isStatusHookTarget func(common.Hash) bool
+	StatusHook         func(common.Hash) TxStatus
 }
 
 func (odr *testOdr) Database() ethdb.Database {
@@ -70,12 +77,18 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error {
 	if odr.disable {
 		return ErrOdrDisabled
 	}
+	var nostore bool
 	switch req := req.(type) {
 	case *BlockRequest:
+		if odr.isBlockHookTarget != nil && odr.isBlockHookTarget(req.Hash) && odr.blockHook != nil {
+			req.Rlp = odr.blockHook(req.Hash)
+			nostore = true
+		} else {
 			number := rawdb.ReadHeaderNumber(odr.sdb, req.Hash)
 			if number != nil {
 				req.Rlp = rawdb.ReadBodyRLP(odr.sdb, req.Hash, *number)
 			}
+		}
 	case *ReceiptsRequest:
 		number := rawdb.ReadHeaderNumber(odr.sdb, req.Hash)
 		if number != nil {
@@ -88,8 +101,24 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error {
 		req.Proof = nodes
 	case *CodeRequest:
 		req.Data, _ = odr.sdb.Get(req.Hash[:])
+	case *TxStatusRequest:
+		var status []TxStatus
+		for _, hash := range req.Hashes {
+			if odr.isStatusHookTarget != nil && odr.isStatusHookTarget(hash) && odr.StatusHook != nil {
+				status = append(status, odr.StatusHook(hash))
+			} else {
+				if tx, blockHash, blockNumber, txIndex := rawdb.ReadTransaction(odr.sdb, hash); tx != nil {
+					status = append(status, TxStatus{Status: core.TxStatusIncluded, Lookup: &rawdb.LegacyTxLookupEntry{BlockHash: blockHash, BlockIndex: blockNumber, Index: txIndex}})
+				} else {
+					status = append(status, TxStatus{Status: core.TxStatusUnknown})
 				}
+			}
+		}
+		req.Status = status
+	}
+	if !nostore {
 		req.StoreResult(odr.ldb)
+	}
 	return nil
 }
 

light/odr_util.go

@@ -229,6 +229,15 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi
 	}
 }
 
+// GetTransactionStatus retrieves the status of a batch of transaction.
+func GetTransactionStatus(ctx context.Context, odr OdrBackend, hashes []common.Hash) ([]TxStatus, error) {
+	r := &TxStatusRequest{Hashes: hashes}
+	if err := odr.Retrieve(ctx, r); err != nil {
+		return nil, err
+	}
+	return r.Status, nil
+}
+
 // GetTransaction retrieves a canonical transaction by hash and also returns its position in the chain
 func GetTransaction(ctx context.Context, odr OdrBackend, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
 	r := &TxStatusRequest{Hashes: []common.Hash{txHash}}

light/txpool_test.go

@@ -18,8 +18,11 @@ package light
 
 import (
 	"context"
+	"errors"
 	"math"
 	"math/big"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -33,26 +36,29 @@ import (
 )
 
 type testTxRelay struct {
-	send, discard, mined chan int
+	send    chan []*types.Transaction
+	discard chan []common.Hash
+
+	sendHook func() error
 }
 
-func (self *testTxRelay) Send(txs types.Transactions) {
+func (t *testTxRelay) Send(txs types.Transactions) error {
-	self.send <- len(txs)
+	t.send <- txs
-}
 
-func (self *testTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
+	if t.sendHook != nil {
-	m := len(mined)
+		return t.sendHook()
-	if m != 0 {
-		self.mined <- m
 	}
+	return nil
 }
 
-func (self *testTxRelay) Discard(hashes []common.Hash) {
+func (t *testTxRelay) Discard(hashes []common.Hash) {
-	self.discard <- len(hashes)
+	t.discard <- hashes
 }
 
-const poolTestTxs = 1000
+const (
-const poolTestBlocks = 100
+	poolTestTxs    = 1000
+	poolTestBlocks = 100
+)
 
 // test tx 0..n-1
 var testTx [poolTestTxs]*types.Transaction
@@ -68,76 +74,351 @@ func minedTx(i int) int {
 }
 
 func txPoolTestChainGen(i int, block *core.BlockGen) {
-	s := minedTx(i)
+	low, high := minedTx(i), minedTx(i+1)
-	e := minedTx(i + 1)
+	for i := low; i < high; i++ {
-	for i := s; i < e; i++ {
 		block.AddTx(testTx[i])
 	}
 }
 
 func TestTxPool(t *testing.T) {
+	var (
+		serverDB = rawdb.NewMemoryDatabase()
+		clientDB = rawdb.NewMemoryDatabase()
+		gspec    = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
+		genesis  = gspec.MustCommit(serverDB)
+		txmap    = make(map[common.Hash]int)
+	)
+	// Initialize transactions
 	for i := range testTx {
 		testTx[i], _ = types.SignTx(types.NewTransaction(uint64(i), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+		txmap[testTx[i].Hash()] = i
 	}
-
+	// Initialize server side.
-	var (
+	blockchain, _ := core.NewBlockChain(serverDB, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
-		sdb     = rawdb.NewMemoryDatabase()
+	gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), serverDB, poolTestBlocks, txPoolTestChainGen)
-		ldb     = rawdb.NewMemoryDatabase()
-		gspec   = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
-		genesis = gspec.MustCommit(sdb)
-	)
-	gspec.MustCommit(ldb)
-	// Assemble the test environment
-	blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
-	gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, poolTestBlocks, txPoolTestChainGen)
 	if _, err := blockchain.InsertChain(gchain); err != nil {
 		panic(err)
 	}
-
+	// Initialize client side.
-	odr := &testOdr{sdb: sdb, ldb: ldb, indexerConfig: TestClientIndexerConfig}
 	relay := &testTxRelay{
-		send:    make(chan int, 1),
+		send:    make(chan []*types.Transaction, 1),
-		discard: make(chan int, 1),
+		discard: make(chan []common.Hash, 1),
-		mined:   make(chan int, 1),
 	}
+	relayCh, discardCh := make(chan error, 1), make(chan error, 1)
+	go func() {
+		received, discarded := make(map[common.Hash]struct{}), make(map[common.Hash]struct{})
+		for {
+			select {
+			case txs := <-relay.send:
+				for _, tx := range txs {
+					if _, exist := txmap[tx.Hash()]; !exist {
+						relayCh <- errors.New("unexpected transaction")
+					}
+					received[tx.Hash()] = struct{}{}
+				}
+				if len(received) == len(testTx) {
+					relayCh <- nil
+				}
+			case hashes := <-relay.discard:
+				for _, h := range hashes {
+					if _, exist := txmap[h]; !exist {
+						discardCh <- errors.New("unexpected transaction")
+					}
+					discarded[h] = struct{}{}
+				}
+				if len(discarded) == len(testTx) {
+					discardCh <- nil
+				}
+			}
+		}
+	}()
+
+	gspec.MustCommit(clientDB)
+	odr := &testOdr{sdb: serverDB, ldb: clientDB, indexerConfig: TestClientIndexerConfig}
+
+	// Register some hooks for various attack testing
+	var statusLock sync.Mutex
+	statusMark := make(map[int]bool)
+	statusCounter := make(map[int]int32)
+	odr.isStatusHookTarget = func(hash common.Hash) bool {
+		statusLock.Lock()
+		defer statusLock.Unlock()
+
+		txIndex := txmap[hash]
+
+		var target bool
+		if txIndex >= 20 && txIndex <= 50 {
+			if !statusMark[txIndex] {
+				statusMark[txIndex] = true
+				target = true
+			}
+		} else if txIndex >= 80 && txIndex <= 100 {
+			if statusCounter[txIndex] <= 10 {
+				statusCounter[txIndex] += 1
+				target = true
+			}
+		}
+		return target
+	}
+	odr.StatusHook = func(hash common.Hash) TxStatus {
+		txIndex := txmap[hash]
+
+		if txIndex >= 20 && txIndex <= 50 {
+			// Respond with a fake status information, expect client
+			// can recover from this attack.
+			return TxStatus{Status: core.TxStatusIncluded, Lookup: &rawdb.LegacyTxLookupEntry{BlockIndex: 0, BlockHash: common.HexToHash("deadbeef")}}
+		} else {
+			// Always respond with unknown status for the first 10 requests,
+			// force light client to resend transaction.
+			return TxStatus{Status: core.TxStatusUnknown}
+		}
+	}
+	var fakeBlock int32
+	odr.isBlockHookTarget = func(hash common.Hash) bool {
+		if hash == gchain[50].Hash() && atomic.LoadInt32(&fakeBlock) <= 5 {
+			atomic.AddInt32(&fakeBlock, 1)
+			return true
+		}
+		return false
+	}
+	odr.blockHook = func(hash common.Hash) []byte {
+		return []byte{0x00, 0x01, 0x02}
+	}
+
 	lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker())
-	txPermanent = 50
+	txPermanent, statusQueryResendDelay = 50, 10*time.Millisecond
-	pool := NewTxPool(params.TestChainConfig, lightchain, relay)
+
+	pool := NewTxPool(core.TxPoolConfig{}, params.TestChainConfig, lightchain, relay)
+	defer pool.Stop()
+
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 	defer cancel()
 
-	for ii, block := range gchain {
+	tail, _ := core.GenerateChain(params.TestChainConfig, gchain[len(gchain)-1], ethash.NewFaker(), serverDB, int(txPermanent), nil)
-		i := ii + 1
-		s := sentTx(i - 1)
-		e := sentTx(i)
-		for i := s; i < e; i++ {
-			pool.Add(ctx, testTx[i])
-			got := <-relay.send
-			exp := 1
-			if got != exp {
-				t.Errorf("relay.Send expected len = %d, got %d", exp, got)
-			}
-		}
 
+	for i, block := range append(gchain, tail...) {
+		low, high := sentTx(i), sentTx(i+1)
+		if high <= len(testTx) && low < high {
+			pool.AddBatch(ctx, testTx[low:high])
+		}
 		if _, err := lightchain.InsertHeaderChain([]*types.Header{block.Header()}, 1); err != nil {
 			panic(err)
 		}
-
+		time.Sleep(10 * time.Millisecond) // Give mainloop enough time to process all events.
-		got := <-relay.mined
-		exp := minedTx(i) - minedTx(i-1)
-		if got != exp {
-			t.Errorf("relay.NewHead expected len(mined) = %d, got %d", exp, got)
 	}
-
+	for _, ch := range []chan error{relayCh, discardCh} {
-		exp = 0
+		select {
-		if i > int(txPermanent)+1 {
+		case err := <-ch:
-			exp = minedTx(i-int(txPermanent)-1) - minedTx(i-int(txPermanent)-2)
+			if err != nil {
+				t.Fatalf("Unexpeated error %v", err)
 			}
-		if exp != 0 {
+		case <-time.NewTimer(5 * time.Second).C:
-			got = <-relay.discard
+			t.Fatalf("timeout")
-			if got != exp {
-				t.Errorf("relay.Discard expected len = %d, got %d", exp, got)
 		}
 	}
+	// Check the integrity in database.
+	for _, tx := range testTx {
+		dbtx, _, _, _ := rawdb.ReadTransaction(clientDB, tx.Hash())
+		if dbtx == nil {
+			t.Fatalf("Transaction %v(index=%d) not found, expect to find in the database", tx.Hash().Hex(), txmap[tx.Hash()])
+		}
+		receipt, _, _, _ := rawdb.ReadReceipt(clientDB, tx.Hash(), lightchain.Config())
+		if receipt == nil {
+			t.Fatalf("Receipt %v not found, expect to find in the database", tx.Hash().Hex())
+		}
+	}
+}
+
+func TestResend(t *testing.T) {
+	var (
+		serverDB = rawdb.NewMemoryDatabase()
+		clientDB = rawdb.NewMemoryDatabase()
+		gspec    = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
+	)
+	// Initialize transactions
+	for i := uint64(0); i < 3; i++ {
+		testTx[i], _ = types.SignTx(types.NewTransaction(i, acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	}
+	// Initialize client side.
+	relay := &testTxRelay{
+		send:    make(chan []*types.Transaction, 1),
+		discard: make(chan []common.Hash, 1),
+	}
+	// Disable relay functionality.
+	relay.sendHook = func() error {
+		return errors.New("test error")
+	}
+	go func() {
+		for {
+			select {
+			case <-relay.send:
+			case <-relay.discard:
+			}
+		}
+	}()
+
+	gspec.MustCommit(clientDB)
+	odr := &testOdr{sdb: serverDB, ldb: clientDB, indexerConfig: TestClientIndexerConfig}
+	lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker())
+
+	pool := NewTxPool(core.TxPoolConfig{}, params.TestChainConfig, lightchain, relay)
+	defer pool.Stop()
+
+	pool.AddBatch(context.Background(), testTx[:3])
+
+	// Resend the transaction with higher transfer value.
+	for i := uint64(0); i < 3; i++ {
+		newtx, _ := types.SignTx(types.NewTransaction(i, acc1Addr, big.NewInt(20000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+		pool.Add(context.Background(), newtx)
+
+		p, _ := pool.GetPendingTransaction(newtx.Hash())
+		if p == nil {
+			t.Fatalf("new transaction should be included")
+		}
+		p, _ = pool.GetPendingTransaction(testTx[i].Hash())
+		if p != nil {
+			t.Fatalf("old transaction should be discarded")
+		}
+	}
+}
+
+func TestInvalidTransaction(t *testing.T) {
+	var (
+		serverDB = rawdb.NewMemoryDatabase()
+		clientDB = rawdb.NewMemoryDatabase()
+		gspec    = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
+	)
+	// Initialize client side.
+	relay := &testTxRelay{
+		send:    make(chan []*types.Transaction, 1),
+		discard: make(chan []common.Hash, 1),
+	}
+	go func() {
+		for {
+			select {
+			case <-relay.send:
+			case <-relay.discard:
+			}
+		}
+	}()
+
+	gspec.MustCommit(clientDB)
+	odr := &testOdr{sdb: serverDB, ldb: clientDB, indexerConfig: TestClientIndexerConfig}
+	lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker())
+
+	pool := NewTxPool(core.TxPoolConfig{}, params.TestChainConfig, lightchain, relay)
+	defer pool.Stop()
+
+	// Duplicated transaction
+	tx, _ := types.SignTx(types.NewTransaction(0, acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	pool.Add(context.Background(), tx)
+	if err := pool.Add(context.Background(), tx); err != errDuplicatedTransaction {
+		t.Fatalf("duplicated transaction expected, %v", err)
+	}
+	// Invalid nonce
+	tx, _ = types.SignTx(types.NewTransaction(2, acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	if err := pool.Add(context.Background(), tx); err != errInvalidNonce {
+		t.Fatalf("invalid nonce error expected, %v", err)
+	}
+	// Not enough balance
+	tx, _ = types.SignTx(types.NewTransaction(0, acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, acc2Key)
+	if err := pool.Add(context.Background(), tx); err != core.ErrInsufficientFunds {
+		t.Fatalf("insufficient funds error expected, %v", err)
+	}
+}
+
+func TestPendingNonce(t *testing.T) {
+	var (
+		serverDB = rawdb.NewMemoryDatabase()
+		clientDB = rawdb.NewMemoryDatabase()
+		gspec    = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
+	)
+	// Initialize transactions
+	for i := range testTx {
+		testTx[i], _ = types.SignTx(types.NewTransaction(uint64(i), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	}
+	// Initialize client side.
+	relay := &testTxRelay{
+		send:    make(chan []*types.Transaction, 1),
+		discard: make(chan []common.Hash, 1),
+	}
+	closeCh := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-relay.send:
+			case <-relay.discard:
+			case <-closeCh:
+			}
+		}
+	}()
+
+	gspec.MustCommit(clientDB)
+	odr := &testOdr{sdb: serverDB, ldb: clientDB, indexerConfig: TestClientIndexerConfig}
+	lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker())
+
+	pool := NewTxPool(core.TxPoolConfig{}, params.TestChainConfig, lightchain, relay)
+	defer pool.Stop()
+
+	pool.AddBatch(context.Background(), testTx[:])
+
+	nonce, err := pool.GetNonce(context.Background(), testBankAddress)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	if nonce != uint64(len(testTx)) {
+		t.Fatalf("nonce mismatch, want %v, have %v", len(testTx), nonce)
+	}
+}
+
+func TestPendingTransaction(t *testing.T) {
+	var (
+		serverDB = rawdb.NewMemoryDatabase()
+		clientDB = rawdb.NewMemoryDatabase()
+		gspec    = core.Genesis{Alloc: core.GenesisAlloc{testBankAddress: {Balance: testBankFunds}}}
+	)
+	// Initialize transactions
+	for i := range testTx[:5] {
+		testTx[i], _ = types.SignTx(types.NewTransaction(uint64(i), acc1Addr, big.NewInt(10000), params.TxGas, nil, nil), types.HomesteadSigner{}, testBankKey)
+	}
+	// Initialize client side.
+	relay := &testTxRelay{
+		send:    make(chan []*types.Transaction, 1),
+		discard: make(chan []common.Hash, 1),
+	}
+	relay.sendHook = func() error {
+		return errors.New("reject relay transaction")
+	}
+	closeCh := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-relay.send:
+			case <-relay.discard:
+			case <-closeCh:
+			}
+		}
+	}()
+
+	gspec.MustCommit(clientDB)
+	odr := &testOdr{sdb: serverDB, ldb: clientDB, indexerConfig: TestClientIndexerConfig}
+	lightchain, _ := NewLightChain(odr, params.TestChainConfig, ethash.NewFullFaker())
+
+	pool := NewTxPool(core.TxPoolConfig{}, params.TestChainConfig, lightchain, relay)
+	defer pool.Stop()
+
+	pool.AddBatch(context.Background(), testTx[:5])
+
+	txs, _ := pool.GetAllPendingTransactions()
+	if len(txs) != 5 {
+		t.Fatalf("Pending transaction number mismatch, want %d, have %d", 5, len(txs))
+	}
+
+	tx, _ := pool.GetPendingTransaction(testTx[0].Hash())
+	if tx == nil {
+		t.Fatalf("Expect pending transaction exists, but not found")
+	}
+	if tx.Hash() != testTx[0].Hash() {
+		t.Fatalf("Pending transaction mismatch, want %s, have %s", testTx[0].Hash().Hex(), tx.Hash().Hex())
 	}
 }