les: implement GetTxStatusMsg and SendTxV2Msg

core/database_util.go

@@ -74,9 +74,9 @@ var (
 	preimageHitCounter = metrics.NewCounter("db/preimage/hits")
 )
 
-// txLookupEntry is a positional metadata to help looking up the data content of
+// TxLookupEntry is a positional metadata to help looking up the data content of
 // a transaction or receipt given only its hash.
-type txLookupEntry struct {
+type TxLookupEntry struct {
 	BlockHash  common.Hash
 	BlockIndex uint64
 	Index      uint64
@@ -260,7 +260,7 @@ func GetTxLookupEntry(db DatabaseReader, hash common.Hash) (common.Hash, uint64,
 		return common.Hash{}, 0, 0
 	}
 	// Parse and return the contents of the lookup entry
-	var entry txLookupEntry
+	var entry TxLookupEntry
 	if err := rlp.DecodeBytes(data, &entry); err != nil {
 		log.Error("Invalid lookup entry RLP", "hash", hash, "err", err)
 		return common.Hash{}, 0, 0
@@ -296,7 +296,7 @@ func GetTransaction(db DatabaseReader, hash common.Hash) (*types.Transaction, co
 	if len(data) == 0 {
 		return nil, common.Hash{}, 0, 0
 	}
-	var entry txLookupEntry
+	var entry TxLookupEntry
 	if err := rlp.DecodeBytes(data, &entry); err != nil {
 		return nil, common.Hash{}, 0, 0
 	}
@@ -464,7 +464,7 @@ func WriteBlockReceipts(db ethdb.Putter, hash common.Hash, number uint64, receip
 func WriteTxLookupEntries(db ethdb.Putter, block *types.Block) error {
 	// Iterate over each transaction and encode its metadata
 	for i, tx := range block.Transactions() {
-		entry := txLookupEntry{
+		entry := TxLookupEntry{
 			BlockHash:  block.Hash(),
 			BlockIndex: block.NumberU64(),
 			Index:      uint64(i),

core/tx_list.go

@@ -384,13 +384,13 @@ func (h *priceHeap) Pop() interface{} {
 // txPricedList is a price-sorted heap to allow operating on transactions pool
 // contents in a price-incrementing way.
 type txPricedList struct {
-	all    *map[common.Hash]*types.Transaction // Pointer to the map of all transactions
+	all    *map[common.Hash]txLookupRec // Pointer to the map of all transactions
 	items  *priceHeap                   // Heap of prices of all the stored transactions
 	stales int                          // Number of stale price points to (re-heap trigger)
 }
 
 // newTxPricedList creates a new price-sorted transaction heap.
-func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList {
+func newTxPricedList(all *map[common.Hash]txLookupRec) *txPricedList {
 	return &txPricedList{
 		all:   all,
 		items: new(priceHeap),
@@ -416,7 +416,7 @@ func (l *txPricedList) Removed() {
 
 	l.stales, l.items = 0, &reheap
 	for _, tx := range *l.all {
-		*l.items = append(*l.items, tx)
+		*l.items = append(*l.items, tx.tx)
 	}
 	heap.Init(l.items)
 }

core/tx_pool.go

@@ -195,7 +195,7 @@ type TxPool struct {
 	pending map[common.Address]*txList   // All currently processable transactions
 	queue   map[common.Address]*txList   // Queued but non-processable transactions
 	beats   map[common.Address]time.Time // Last heartbeat from each known account
-	all     map[common.Hash]*types.Transaction // All transactions to allow lookups
+	all     map[common.Hash]txLookupRec  // All transactions to allow lookups
 	priced  *txPricedList                // All transactions sorted by price
 
 	wg sync.WaitGroup // for shutdown sync
@@ -203,6 +203,11 @@ type TxPool struct {
 	homestead bool
 }
 
+type txLookupRec struct {
+	tx      *types.Transaction
+	pending bool
+}
+
 // NewTxPool creates a new transaction pool to gather, sort and filter inbound
 // trnsactions from the network.
 func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain blockChain) *TxPool {
@@ -218,7 +223,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
 		pending:     make(map[common.Address]*txList),
 		queue:       make(map[common.Address]*txList),
 		beats:       make(map[common.Address]time.Time),
-		all:         make(map[common.Hash]*types.Transaction),
+		all:         make(map[common.Hash]txLookupRec),
 		chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
 		gasPrice:    new(big.Int).SetUint64(config.PriceLimit),
 	}
@@ -594,7 +599,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
 func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 	// If the transaction is already known, discard it
 	hash := tx.Hash()
-	if pool.all[hash] != nil {
+	if _, ok := pool.all[hash]; ok {
 		log.Trace("Discarding already known transaction", "hash", hash)
 		return false, fmt.Errorf("known transaction: %x", hash)
 	}
@@ -635,7 +640,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 			pool.priced.Removed()
 			pendingReplaceCounter.Inc(1)
 		}
-		pool.all[tx.Hash()] = tx
+		pool.all[tx.Hash()] = txLookupRec{tx, false}
 		pool.priced.Put(tx)
 		pool.journalTx(from, tx)
 
@@ -678,7 +683,7 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
 		pool.priced.Removed()
 		queuedReplaceCounter.Inc(1)
 	}
-	pool.all[hash] = tx
+	pool.all[hash] = txLookupRec{tx, false}
 	pool.priced.Put(tx)
 	return old != nil, nil
 }
@@ -721,10 +726,13 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 
 		pendingReplaceCounter.Inc(1)
 	}
+	if pool.all[hash].tx == nil {
 		// Failsafe to work around direct pending inserts (tests)
-	if pool.all[hash] == nil {
+		pool.all[hash] = txLookupRec{tx, true}
-		pool.all[hash] = tx
 		pool.priced.Put(tx)
+	} else {
+		// set pending flag to true
+		pool.all[hash] = txLookupRec{tx, true}
 	}
 	// Set the potentially new pending nonce and notify any subsystems of the new tx
 	pool.beats[addr] = time.Now()
@@ -750,14 +758,16 @@ func (pool *TxPool) AddRemote(tx *types.Transaction) error {
 // marking the senders as a local ones in the mean time, ensuring they go around
 // the local pricing constraints.
 func (pool *TxPool) AddLocals(txs []*types.Transaction) error {
-	return pool.addTxs(txs, !pool.config.NoLocals)
+	pool.addTxs(txs, !pool.config.NoLocals)
+	return nil
 }
 
 // AddRemotes enqueues a batch of transactions into the pool if they are valid.
 // If the senders are not among the locally tracked ones, full pricing constraints
 // will apply.
 func (pool *TxPool) AddRemotes(txs []*types.Transaction) error {
-	return pool.addTxs(txs, false)
+	pool.addTxs(txs, false)
+	return nil
 }
 
 // addTx enqueues a single transaction into the pool if it is valid.
@@ -779,7 +789,7 @@ func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
 }
 
 // addTxs attempts to queue a batch of transactions if they are valid.
-func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error {
+func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) []error {
 	pool.mu.Lock()
 	defer pool.mu.Unlock()
 
@@ -788,11 +798,13 @@ func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) error {
 
 // addTxsLocked attempts to queue a batch of transactions if they are valid,
 // whilst assuming the transaction pool lock is already held.
-func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) error {
+func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) []error {
 	// Add the batch of transaction, tracking the accepted ones
 	dirty := make(map[common.Address]struct{})
-	for _, tx := range txs {
+	txErr := make([]error, len(txs))
-		if replace, err := pool.add(tx, local); err == nil {
+	for i, tx := range txs {
+		var replace bool
+		if replace, txErr[i] = pool.add(tx, local); txErr[i] == nil {
 			if !replace {
 				from, _ := types.Sender(pool.signer, tx) // already validated
 				dirty[from] = struct{}{}
@@ -807,7 +819,58 @@ func (pool *TxPool) addTxsLocked(txs []*types.Transaction, local bool) error {
 		}
 		pool.promoteExecutables(addrs)
 	}
-	return nil
+	return txErr
+}
+
+// TxStatusData is returned by AddOrGetTxStatus for each transaction
+type TxStatusData struct {
+	Status uint
+	Data   []byte
+}
+
+const (
+	TxStatusUnknown = iota
+	TxStatusQueued
+	TxStatusPending
+	TxStatusIncluded // Data contains a TxChainPos struct
+	TxStatusError    // Data contains the error string
+)
+
+// AddOrGetTxStatus returns the status (unknown/pending/queued) of a batch of transactions
+// identified by their hashes in txHashes. Optionally the transactions themselves can be
+// passed too in txs, in which case the function will try adding the previously unknown ones
+// to the pool. If a new transaction cannot be added, TxStatusError is returned. Adding already
+// known transactions will return their previous status.
+// If txs is specified, txHashes is still required and has to match the transactions in txs.
+
+// Note: TxStatusIncluded is never returned by this function since the pool does not track
+// mined transactions. Included status can be checked by the caller (as it happens in the
+// LES protocol manager)
+func (pool *TxPool) AddOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []TxStatusData {
+	status := make([]TxStatusData, len(txHashes))
+	if txs != nil {
+		if len(txs) != len(txHashes) {
+			panic(nil)
+		}
+		txErr := pool.addTxs(txs, false)
+		for i, err := range txErr {
+			if err != nil {
+				status[i] = TxStatusData{TxStatusError, ([]byte)(err.Error())}
+			}
+		}
+	}
+
+	for i, hash := range txHashes {
+		r, ok := pool.all[hash]
+		if ok {
+			if r.pending {
+				status[i] = TxStatusData{TxStatusPending, nil}
+			} else {
+				status[i] = TxStatusData{TxStatusQueued, nil}
+			}
+		}
+	}
+	return status
 }
 
 // Get returns a transaction if it is contained in the pool
@@ -816,17 +879,18 @@ func (pool *TxPool) Get(hash common.Hash) *types.Transaction {
 	pool.mu.RLock()
 	defer pool.mu.RUnlock()
 
-	return pool.all[hash]
+	return pool.all[hash].tx
 }
 
 // removeTx removes a single transaction from the queue, moving all subsequent
 // transactions back to the future queue.
 func (pool *TxPool) removeTx(hash common.Hash) {
 	// Fetch the transaction we wish to delete
-	tx, ok := pool.all[hash]
+	txl, ok := pool.all[hash]
 	if !ok {
 		return
 	}
+	tx := txl.tx
 	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
 
 	// Remove it from the list of known transactions

les/handler.go

@@ -59,6 +59,7 @@ const (
 	MaxProofsFetch    = 64  // Amount of merkle proofs to be fetched per retrieval request
 	MaxPPTProofsFetch = 64  // Amount of merkle proofs to be fetched per retrieval request
 	MaxTxSend         = 64  // Amount of transactions to be send per request
+	MaxTxStatus       = 256 // Amount of transactions to queried per request
 
 	disableClientRemovePeer = false
 )
@@ -88,8 +89,7 @@ type BlockChain interface {
 }
 
 type txPool interface {
-	// AddRemotes should add the given transactions to the pool.
+	AddOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []core.TxStatusData
-	AddRemotes([]*types.Transaction) error
 }
 
 type ProtocolManager struct {
@@ -318,7 +318,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	}
 }
 
-var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsV1Msg, SendTxMsg, GetHeaderProofsMsg, GetProofsV2Msg, GetPPTProofsMsg}
+var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsV1Msg, SendTxMsg, SendTxV2Msg, GetTxStatusMsg, GetHeaderProofsMsg, GetProofsV2Msg, GetPPTProofsMsg}
 
 // handleMsg is invoked whenever an inbound message is received from a remote
 // peer. The remote connection is torn down upon returning any error.
@@ -972,7 +972,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 
 	case SendTxMsg:
 		if pm.txpool == nil {
-			return errResp(ErrUnexpectedResponse, "")
+			return errResp(ErrRequestRejected, "")
 		}
 		// Transactions arrived, parse all of them and deliver to the pool
 		var txs []*types.Transaction
@@ -984,13 +984,82 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrRequestRejected, "")
 		}
 
-		if err := pm.txpool.AddRemotes(txs); err != nil {
+		txHashes := make([]common.Hash, len(txs))
-			return errResp(ErrUnexpectedResponse, "msg: %v", err)
+		for i, tx := range txs {
+			txHashes[i] = tx.Hash()
 		}
+		pm.addOrGetTxStatus(txs, txHashes)
 
 		_, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 
+	case SendTxV2Msg:
+		if pm.txpool == nil {
+			return errResp(ErrRequestRejected, "")
+		}
+		// Transactions arrived, parse all of them and deliver to the pool
+		var req struct {
+			ReqID uint64
+			Txs   []*types.Transaction
+		}
+		if err := msg.Decode(&req); err != nil {
+			return errResp(ErrDecode, "msg %v: %v", msg, err)
+		}
+		reqCnt := len(req.Txs)
+		if reject(uint64(reqCnt), MaxTxSend) {
+			return errResp(ErrRequestRejected, "")
+		}
+
+		txHashes := make([]common.Hash, len(req.Txs))
+		for i, tx := range req.Txs {
+			txHashes[i] = tx.Hash()
+		}
+
+		res := pm.addOrGetTxStatus(req.Txs, txHashes)
+
+		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
+		return p.SendTxStatus(req.ReqID, bv, res)
+
+	case GetTxStatusMsg:
+		if pm.txpool == nil {
+			return errResp(ErrUnexpectedResponse, "")
+		}
+		// Transactions arrived, parse all of them and deliver to the pool
+		var req struct {
+			ReqID    uint64
+			TxHashes []common.Hash
+		}
+		if err := msg.Decode(&req); err != nil {
+			return errResp(ErrDecode, "msg %v: %v", msg, err)
+		}
+		reqCnt := len(req.TxHashes)
+		if reject(uint64(reqCnt), MaxTxStatus) {
+			return errResp(ErrRequestRejected, "")
+		}
+
+		res := pm.addOrGetTxStatus(nil, req.TxHashes)
+
+		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
+		return p.SendTxStatus(req.ReqID, bv, res)
+
+	case TxStatusMsg:
+		if pm.odr == nil {
+			return errResp(ErrUnexpectedResponse, "")
+		}
+
+		p.Log().Trace("Received tx status response")
+		var resp struct {
+			ReqID, BV uint64
+			Status    []core.TxStatusData
+		}
+		if err := msg.Decode(&resp); err != nil {
+			return errResp(ErrDecode, "msg %v: %v", msg, err)
+		}
+
+		p.fcServer.GotReply(resp.ReqID, resp.BV)
+
 	default:
 		p.Log().Trace("Received unknown message", "code", msg.Code)
 		return errResp(ErrInvalidMsgCode, "%v", msg.Code)
@@ -1034,6 +1103,21 @@ func (pm *ProtocolManager) getPPTAuxData(req PPTReq) []byte {
 	return nil
 }
 
+func (pm *ProtocolManager) addOrGetTxStatus(txs []*types.Transaction, txHashes []common.Hash) []core.TxStatusData {
+	status := pm.txpool.AddOrGetTxStatus(txs, txHashes)
+	for i, _ := range status {
+		blockHash, blockNum, txIndex := core.GetTxLookupEntry(pm.chainDb, txHashes[i])
+		if blockHash != (common.Hash{}) {
+			enc, err := rlp.EncodeToBytes(core.TxLookupEntry{BlockHash: blockHash, BlockIndex: blockNum, Index: txIndex})
+			if err != nil {
+				panic(err)
+			}
+			status[i] = core.TxStatusData{Status: core.TxStatusIncluded, Data: enc}
+		}
+	}
+	return status
+}
+
 // NodeInfo retrieves some protocol metadata about the running host node.
 func (self *ProtocolManager) NodeInfo() *eth.EthNodeInfo {
 	return &eth.EthNodeInfo{

les/peer.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/les/flowcontrol"
@@ -231,6 +232,11 @@ func (p *peer) SendPPTProofs(reqID, bv uint64, resp PPTResps) error {
 	return sendResponse(p.rw, PPTProofsMsg, reqID, bv, resp)
 }
 
+// SendTxStatus sends a batch of transaction status records, corresponding to the ones requested.
+func (p *peer) SendTxStatus(reqID, bv uint64, status []core.TxStatusData) error {
+	return sendResponse(p.rw, TxStatusMsg, reqID, bv, status)
+}
+
 // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the
 // specified header query, based on the hash of an origin block.
 func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amount int, skip int, reverse bool) error {
@@ -303,7 +309,14 @@ func (p *peer) RequestPPTProofs(reqID, cost uint64, reqs []PPTReq) error {
 
 func (p *peer) SendTxs(reqID, cost uint64, txs types.Transactions) error {
 	p.Log().Debug("Fetching batch of transactions", "count", len(txs))
-	return p2p.Send(p.rw, SendTxMsg, txs)
+	switch p.version {
+	case lpv1:
+		return p2p.Send(p.rw, SendTxMsg, txs) // old message format does not include reqID
+	case lpv2:
+		return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs)
+	default:
+		panic(nil)
+	}
 }
 
 type keyValueEntry struct {

les/protocol.go

@@ -45,7 +45,7 @@ var (
 )
 
 // Number of implemented message corresponding to different protocol versions.
-var ProtocolLengths = map[uint]uint64{lpv1: 15, lpv2: 19}
+var ProtocolLengths = map[uint]uint64{lpv1: 15, lpv2: 22}
 
 const (
 	NetworkId          = 1
@@ -75,6 +75,9 @@ const (
 	ProofsV2Msg     = 0x10
 	GetPPTProofsMsg = 0x11
 	PPTProofsMsg    = 0x12
+	SendTxV2Msg     = 0x13
+	GetTxStatusMsg  = 0x14
+	TxStatusMsg     = 0x15
 )
 
 type errCode int