les, light: resend if the transaction is not retrieved back

Author: rjl493456442

les/client.go

@@ -122,7 +122,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*LightEthereum, error) {
 	leth.retriever = newRetrieveManager(peers, leth.reqDist, leth.serverPool.getTimeout)
 	leth.relay = newLesTxRelay(peers, leth.retriever)
 
-	leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.retriever)
+	leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.peers, leth.retriever)
 	leth.chtIndexer = light.NewChtIndexer(chainDb, leth.odr, params.CHTFrequency, params.HelperTrieConfirmations, config.LightNoPrune)
 	leth.bloomTrieIndexer = light.NewBloomTrieIndexer(chainDb, leth.odr, params.BloomBitsBlocksClient, params.BloomTrieFrequency, config.LightNoPrune)
 	leth.odr.SetIndexers(leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer)

les/odr.go

@@ -18,6 +18,7 @@ package les
 
 import (
 	"context"
+	"sort"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
@@ -31,14 +32,16 @@ type LesOdr struct {
 	db                                         ethdb.Database
 	indexerConfig                              *light.IndexerConfig
 	chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer
+	peers                                      *serverPeerSet
 	retriever                                  *retrieveManager
 	stop                                       chan struct{}
 }
 
-func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, retriever *retrieveManager) *LesOdr {
+func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, peers *serverPeerSet, retriever *retrieveManager) *LesOdr {
 	return &LesOdr{
 		db:            db,
 		indexerConfig: config,
+		peers:         peers,
 		retriever:     retriever,
 		stop:          make(chan struct{}),
 	}
@@ -98,7 +101,101 @@ type Msg struct {
 	Obj     interface{}
 }
 
-// Retrieve tries to fetch an object from the LES network.
+// peerByTxHistory is a heap.Interface implementation which can sort
+// the peerset by transaction history.
+type peerByTxHistory []*serverPeer
+
+func (h peerByTxHistory) Len() int { return len(h) }
+func (h peerByTxHistory) Less(i, j int) bool {
+	if h[i].txHistory == txIndexUnlimited {
+		return false
+	}
+	if h[j].txHistory == txIndexUnlimited {
+		return true
+	}
+	return h[i].txHistory < h[j].txHistory
+}
+func (h peerByTxHistory) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
+
+const (
+	maxTxStatusRetry      = 3 // The maximum retrys will be made for tx status request.
+	maxTxStatusCandidates = 5 // The maximum les servers the tx status requests will be sent to.
+)
+
+// RetrieveTxStatus retrieves the transaction status from the LES network.
+// There is no guarantee in the LES protocol that the mined transaction will
+// be retrieved back for sure because of different reasons(the transaction
+// is unindexed, the malicous server doesn't reply it deliberately, etc).
+// Therefore, unretrieved transactions(UNKNOWN) will receive a certain number
+// of retrys, thus giving a weak guarantee.
+func (odr *LesOdr) RetrieveTxStatus(ctx context.Context, req *light.TxStatusRequest) error {
+	// Sort according to the transaction history supported by the peer and
+	// select the peers with longest history.
+	var (
+		retrys  int
+		peers   []*serverPeer
+		missing = len(req.Hashes)
+		result  = make([]light.TxStatus, len(req.Hashes))
+		canSend = make(map[string]bool)
+	)
+	for _, peer := range odr.peers.allPeers() {
+		if peer.txHistory == txIndexDisabled {
+			continue
+		}
+		peers = append(peers, peer)
+	}
+	sort.Sort(sort.Reverse(peerByTxHistory(peers)))
+	for i := 0; i < maxTxStatusCandidates && i < len(peers); i++ {
+		canSend[peers[i].id] = true
+	}
+	// Send out the request and assemble the result.
+	for {
+		if retrys >= maxTxStatusRetry {
+			break
+		}
+		var (
+			// Deep copy the request, so that the partial result won't be mixed.
+			req     = &TxStatusRequest{Hashes: req.Hashes}
+			id      = genReqID()
+			distreq = &distReq{
+				getCost: func(dp distPeer) uint64 { return req.GetCost(dp.(*serverPeer)) },
+				canSend: func(dp distPeer) bool { return canSend[dp.(*serverPeer).id] },
+				request: func(dp distPeer) func() {
+					p := dp.(*serverPeer)
+					p.fcServer.QueuedRequest(id, req.GetCost(p))
+					delete(canSend, p.id)
+					return func() { req.Request(id, p) }
+				},
+			}
+		)
+		if err := odr.retriever.retrieve(ctx, id, distreq, func(p distPeer, msg *Msg) error { return req.Validate(odr.db, msg) }, odr.stop); err != nil {
+			return err
+		}
+		// Collect the response and assemble them to the final result.
+		// All the response is not verifiable, so always pick the first
+		// one we get.
+		for index, status := range req.Status {
+			if result[index].Status != core.TxStatusUnknown {
+				continue
+			}
+			if status.Status == core.TxStatusUnknown {
+				continue
+			}
+			result[index], missing = status, missing-1
+		}
+		// Abort the procedure if all the status are retrieved
+		if missing == 0 {
+			break
+		}
+		retrys += 1
+	}
+	req.Status = result
+	return nil
+}
+
+// Retrieve tries to fetch an object from the LES network. It's a common API
+// for most of the LES requests except for the TxStatusRequest which needs
+// the addtional retry mechanism.
 // If the network retrieval was successful, it stores the object in local db.
 func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err error) {
 	lreq := LesRequest(req)

les/odr_requests.go

@@ -487,7 +487,7 @@ func (r *TxStatusRequest) GetCost(peer *serverPeer) uint64 {
 
 // CanSend tells if a certain peer is suitable for serving the given request
 func (r *TxStatusRequest) CanSend(peer *serverPeer) bool {
-	return peer.serveTxLookup
+	return true
 }
 
 // Request sends an ODR request to the LES network (implementation of LesOdrRequest)
@@ -496,13 +496,12 @@ func (r *TxStatusRequest) Request(reqID uint64, peer *serverPeer) error {
 	return peer.requestTxStatus(reqID, r.Hashes)
 }
 
-// Valid processes an ODR request reply message from the LES network
+// Validate processes an ODR request reply message from the LES network
 // returns true and stores results in memory if the message was a valid reply
 // to the request (implementation of LesOdrRequest)
 func (r *TxStatusRequest) Validate(db ethdb.Database, msg *Msg) error {
 	log.Debug("Validating transaction status", "count", len(r.Hashes))
 
-	// Ensure we have a correct message with a single block body
 	if msg.MsgType != MsgTxStatus {
 		return errInvalidMessageType
 	}

les/peer.go

@@ -341,7 +341,7 @@ type serverPeer struct {
 	onlyAnnounce            bool   // The flag whether the server sends announcement only.
 	chainSince, chainRecent uint64 // The range of chain server peer can serve.
 	stateSince, stateRecent uint64 // The range of state server peer can serve.
-	serveTxLookup           bool   // The server peer can serve tx lookups.
+	txHistory               uint64 // The length of available tx history, 0 means all, 1 means disabled
 
 	// Advertised checkpoint fields
 	checkpointNumber uint64                   // The block height which the checkpoint is registered.
@@ -634,13 +634,13 @@ func (p *serverPeer) Handshake(genesis common.Hash, forkid forkid.ID, forkFilter
 			if err := recv.get("recentTxLookup", &recentTx); err != nil {
 				return err
 			}
-			// Note: in the current version we only consider the tx index service useful
-			// if it is unlimited. This can be made configurable in the future.
-			p.serveTxLookup = recentTx == txIndexUnlimited
+			p.txHistory = uint64(recentTx)
 		} else {
-			p.serveTxLookup = true
+			// The weak assumption is held here that legacy les server(les2,3)
+			// has unlimited transaction history. The les serving in these legacy
+			// versions is disabled if the transaction is unindexed.
+			p.txHistory = txIndexUnlimited
 		}
-
 		if p.onlyAnnounce && !p.trusted {
 			return errResp(ErrUselessPeer, "peer cannot serve requests")
 		}

les/test_helper.go

@@ -523,7 +523,7 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, callback indexer
 	}
 	dist := newRequestDistributor(speers, clock)
 	rm := newRetrieveManager(speers, dist, func() time.Duration { return time.Millisecond * 500 })
-	odr := NewLesOdr(cdb, light.TestClientIndexerConfig, rm)
+	odr := NewLesOdr(cdb, light.TestClientIndexerConfig, speers, rm)
 
 	sindexers := testIndexers(sdb, nil, light.TestServerIndexerConfig, true)
 	cIndexers := testIndexers(cdb, odr, light.TestClientIndexerConfig, disablePruning)

light/odr.go

@@ -42,6 +42,7 @@ type OdrBackend interface {
 	BloomTrieIndexer() *core.ChainIndexer
 	BloomIndexer() *core.ChainIndexer
 	Retrieve(ctx context.Context, req OdrRequest) error
+	RetrieveTxStatus(ctx context.Context, req *TxStatusRequest) error
 	IndexerConfig() *IndexerConfig
 }
 

light/odr_util.go

@@ -269,10 +269,15 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bit uint, sections []uint
 	return result, nil
 }
 
-// GetTransaction retrieves a canonical transaction by hash and also returns its position in the chain
+// GetTransaction retrieves a canonical transaction by hash and also returns
+// its position in the chain. There is no guarantee in the LES protocol that
+// the mined transaction will be retrieved back for sure because of different
+// reasons(the transaction is unindexed, the malicous server doesn't reply it
+// deliberately, etc). Therefore, unretrieved transactions will receive a certain
+// number of retrys, thus giving a weak guarantee.
 func GetTransaction(ctx context.Context, odr OdrBackend, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
 	r := &TxStatusRequest{Hashes: []common.Hash{txHash}}
-	if err := odr.Retrieve(ctx, r); err != nil || r.Status[0].Status != core.TxStatusIncluded {
+	if err := odr.RetrieveTxStatus(ctx, r); err != nil || r.Status[0].Status != core.TxStatusIncluded {
 		return nil, common.Hash{}, 0, 0, err
 	}
 	pos := r.Status[0].Lookup