ethclient/lightclient: implemented pending transaction tracking

ethclient/lightclient/chain.go

@@ -41,7 +41,7 @@ type canonicalChain struct {
 	lock             sync.Mutex
 	headTracker      *light.HeadTracker
 	blocksAndHeaders *blocksAndHeaders
-	newHeadCb        func(common.Hash)
+	newHeadCb        func(uint64, common.Hash)
 
 	head, finality *btypes.ExecutionHeader
 	recent         map[uint64]common.Hash // nil until initialized
@@ -51,7 +51,7 @@ type canonicalChain struct {
 	requests       *requestMap[uint64, common.Hash] // requested; neither recent nor cached finalized
 }
 
-func newCanonicalChain(headTracker *light.HeadTracker, blocksAndHeaders *blocksAndHeaders, newHeadCb func(common.Hash)) *canonicalChain {
+func newCanonicalChain(headTracker *light.HeadTracker, blocksAndHeaders *blocksAndHeaders, newHeadCb func(uint64, common.Hash)) *canonicalChain {
 	c := &canonicalChain{
 		headTracker:      headTracker,
 		blocksAndHeaders: blocksAndHeaders,
@@ -75,7 +75,7 @@ func (c *canonicalChain) Process(requester request.Requester, events []request.E
 		head := optimistic.Attested.PayloadHeader
 		c.blocksAndHeaders.addPayloadHeader(head)
 		if c.setHead(head) {
-			c.newHeadCb(head.BlockHash()) // should not block
+			c.newHeadCb(head.BlockNumber(), head.BlockHash()) // should not block
 		}
 	}
 }

ethclient/lightclient/lightclient.go

@@ -74,6 +74,7 @@ func NewClient(clConfig config.LightClientConfig, elConfig *params.ChainConfig,
 	client.txAndReceipts = txAndReceipts
 	client.canonicalChain = canonicalChain
 	client.state = newLightState(rpcClient, canonicalChain, blocksAndHeaders)
+	txAndReceipts.lightState = client.state //TODO init these refs in a nicer way???
 
 	checkpointInit := sync.NewCheckpointInit(committeeChain, clConfig.Checkpoint)
 	forwardSync := sync.NewForwardUpdateSync(committeeChain)
@@ -159,11 +160,16 @@ func (c *Client) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header)
 	return sub, nil
 }
 
-func (c *Client) newHead(hash common.Hash) {
+func (c *Client) newHead(number uint64, hash common.Hash) {
 	go func() {
 		log.Trace("New execution payload header received", "hash", hash)
+		c.txAndReceipts.newHead(number, hash)
 		ctx, cancel := context.WithCancel(context.Background())
 		c.headSubLock.Lock()
+		if len(c.headSubs) == 0 {
+			c.headSubLock.Unlock()
+			return
+		}
 		if c.cancelHeadFetch != nil {
 			c.cancelHeadFetch()
 		}

ethclient/lightclient/state.go

@@ -252,72 +252,9 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
 	if stateRootErr != nil {
 		return nil, nil, stateRootErr
 	}
-	proofReader, err := makeProofReader(proof.AccountProof)
+	if err := s.validateProof(proof, stateRoot, account, storageKeys); err != nil {
-	if err != nil {
 		return nil, nil, err
 	}
-	value, err := trie.VerifyProof(stateRoot, crypto.Keccak256(account.Bytes()), proofReader)
-	if err != nil {
-		return nil, nil, err
-	}
-	if proof.Balance == nil {
-		return nil, nil, errors.New("account balance is nil")
-	}
-	balance, overflow := uint256.FromBig(proof.Balance)
-	if overflow {
-		return nil, nil, errors.New("account balance overflow")
-	}
-	stateAccount := types.StateAccount{
-		Nonce:    proof.Nonce,
-		Balance:  balance,
-		Root:     proof.StorageHash,
-		CodeHash: proof.CodeHash.Bytes(),
-	}
-	enc, _ := rlp.EncodeToBytes(&stateAccount)
-	if !bytes.Equal(enc, value) {
-		return nil, nil, errors.New("account RLP mismatch")
-	}
-	if len(storageKeys) != len(proof.StorageProof) {
-		return nil, nil, errors.New("invalid number of storage proofs")
-	}
-	for i, st := range proof.StorageProof {
-		if proof.StorageHash == types.EmptyRootHash {
-			// no storage trie, expect empty proofs and values
-			if len(st.Proof) != 0 {
-				return nil, nil, errors.New("non-empty storage proof from empty storage")
-			}
-			value, err := stValueBytes(st.Value)
-			if err != nil {
-				return nil, nil, err
-			}
-			if value != nil {
-				return nil, nil, errors.New("non-empty storage value from empty storage")
-			}
-			continue
-		}
-		proofReader, err := makeProofReader(st.Proof)
-		if err != nil {
-			return nil, nil, err
-		}
-		key, err := hexutil.Decode(storageKeys[i])
-		if err != nil {
-			return nil, nil, err
-		}
-		key = common.BytesToHash(key).Bytes() // TODO 32 byte padding needed???
-		value, err := trie.VerifyProof(proof.StorageHash, crypto.Keccak256(key), proofReader)
-		if err != nil {
-			return nil, nil, err
-		}
-		stv, err := stValueBytes(st.Value)
-		if err != nil {
-			return nil, nil, err
-		}
-		enc, _ := rlp.EncodeToBytes(stv)
-		if !bytes.Equal(enc, value) { //TODO check for empty value
-			//log.Info("storage value mismatch", "value", enc, "proven", value)
-			return nil, nil, errors.New("storage value mismatch")
-		}
-	}
 	if getCode {
 		<-codeCh
 		if codeErr != nil {
@@ -329,3 +266,76 @@ func (s *lightState) getProof(ctx context.Context, blockNumber *big.Int, account
 	}
 	return proof, code, nil
 }
+
+func (s *lightState) validateProof(proof *gethclient.AccountResult, stateRoot common.Hash, account common.Address, storageKeys []string) error {
+	// validate account proof
+	proofReader, err := makeProofReader(proof.AccountProof)
+	if err != nil {
+		return err
+	}
+	value, err := trie.VerifyProof(stateRoot, crypto.Keccak256(account.Bytes()), proofReader)
+	if err != nil {
+		return err
+	}
+	if proof.Balance == nil {
+		return errors.New("account balance is nil")
+	}
+	balance, overflow := uint256.FromBig(proof.Balance)
+	if overflow {
+		return errors.New("account balance overflow")
+	}
+	stateAccount := types.StateAccount{
+		Nonce:    proof.Nonce,
+		Balance:  balance,
+		Root:     proof.StorageHash,
+		CodeHash: proof.CodeHash.Bytes(),
+	}
+	enc, _ := rlp.EncodeToBytes(&stateAccount)
+	if !bytes.Equal(enc, value) {
+		return errors.New("account RLP mismatch")
+	}
+
+	// validate storage proofs
+	if len(storageKeys) != len(proof.StorageProof) {
+		return errors.New("invalid number of storage proofs")
+	}
+	for i, st := range proof.StorageProof {
+		if proof.StorageHash == types.EmptyRootHash {
+			// no storage trie, expect empty proofs and values
+			if len(st.Proof) != 0 {
+				return errors.New("non-empty storage proof from empty storage")
+			}
+			value, err := stValueBytes(st.Value)
+			if err != nil {
+				return err
+			}
+			if value != nil {
+				return errors.New("non-empty storage value from empty storage")
+			}
+			continue
+		}
+		proofReader, err := makeProofReader(st.Proof)
+		if err != nil {
+			return err
+		}
+		key, err := hexutil.Decode(storageKeys[i])
+		if err != nil {
+			return err
+		}
+		key = common.BytesToHash(key).Bytes() // TODO 32 byte padding needed???
+		value, err := trie.VerifyProof(proof.StorageHash, crypto.Keccak256(key), proofReader)
+		if err != nil {
+			return err
+		}
+		stv, err := stValueBytes(st.Value)
+		if err != nil {
+			return err
+		}
+		enc, _ := rlp.EncodeToBytes(stv)
+		if !bytes.Equal(enc, value) { //TODO check for empty value
+			//log.Info("storage value mismatch", "value", enc, "proven", value)
+			return errors.New("storage value mismatch")
+		}
+	}
+	return nil
+}

ethclient/lightclient/transactions.go

@@ -22,8 +22,11 @@ import (
 	"encoding/json"
 	"errors"
 	"math/big"
+	"sort"
+	"sync"
 
 	"github.com/ethereum/go-ethereum"
+	btypes "github.com/ethereum/go-ethereum/beacon/types"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/common/lru"
@@ -34,14 +37,23 @@ import (
 	"github.com/ethereum/go-ethereum/trie"
 )
 
+const maxTxAge = 300 // sent transactions are typically remembered for an hour after last seen pending
+
 type txAndReceipts struct {
 	client           *rpc.Client
 	canonicalChain   *canonicalChain
 	blocksAndHeaders *blocksAndHeaders
+	lightState       *lightState
 	elConfig         *params.ChainConfig
+	signer           types.Signer
+
 	receiptsCache    *lru.Cache[common.Hash, types.Receipts]
 	receiptsRequests *requestMap[common.Hash, types.Receipts]
 	txPosCache       *lru.Cache[common.Hash, txInBlock]
+
+	sentTxLock                  sync.Mutex
+	sentTxs                     map[common.Address]senderTxs
+	headCounter, lastHeadNumber uint64
 }
 
 type txInBlock struct {
@@ -50,6 +62,13 @@ type txInBlock struct {
 	index       uint
 }
 
+type sentTx struct {
+	nonce    uint64
+	lastSeen uint64 // headCounter value where the tx has been sent or last seen pending
+}
+
+type senderTxs map[common.Hash]sentTx
+
 func newTxAndReceipts(client *rpc.Client, canonicalChain *canonicalChain, blocksAndHeaders *blocksAndHeaders, elConfig *params.ChainConfig) *txAndReceipts {
 	t := &txAndReceipts{
 		client:           client,
@@ -58,6 +77,8 @@ func newTxAndReceipts(client *rpc.Client, canonicalChain *canonicalChain, blocks
 		elConfig:         elConfig,
 		receiptsCache:    lru.NewCache[common.Hash, types.Receipts](10),
 		txPosCache:       lru.NewCache[common.Hash, txInBlock](10000),
+		sentTxs:          make(map[common.Address]senderTxs),
+		signer:           types.LatestSigner(elConfig),
 	}
 	t.receiptsRequests = newRequestMap[common.Hash, types.Receipts](t.requestBlockReceipts)
 	return t
@@ -71,6 +92,14 @@ func (t *txAndReceipts) getTxByHash(ctx context.Context, txHash common.Hash) (tx
 			}
 		}
 	}
+	var headBlockNumber uint64
+	if head := t.canonicalChain.getHead(); head != nil {
+		headBlockNumber = head.BlockNumber()
+	}
+	return t.getUncachedTxByHash(ctx, txHash, headBlockNumber)
+}
+
+func (t *txAndReceipts) getUncachedTxByHash(ctx context.Context, txHash common.Hash, headBlockNumber uint64) (tx *types.Transaction, isPending bool, err error) {
 	tx, isPending, err = t.requestTxByHash(ctx, txHash)
 	if err == nil && !isPending {
 		receipt, err := t.requestReceiptByTxHash(ctx, txHash)
@@ -80,8 +109,7 @@ func (t *txAndReceipts) getTxByHash(ctx context.Context, txHash common.Hash) (tx
 		if err != nil {
 			return nil, false, err
 		}
-		if head := t.canonicalChain.getHead(); head == nil ||
+		if !receipt.BlockNumber.IsUint64() || headBlockNumber < receipt.BlockNumber.Uint64() {
-			!receipt.BlockNumber.IsUint64() || head.BlockNumber() < receipt.BlockNumber.Uint64() {
 			// consider it pending if it's reported to be included higher than the light chain head
 			isPending = true
 		}
@@ -221,6 +249,8 @@ func (t *txAndReceipts) requestTxByHash(ctx context.Context, hash common.Hash) (
 		return nil, false, ethereum.NotFound
 	} else if _, r, _ := json.tx.RawSignatureValues(); r == nil {
 		return nil, false, errors.New("server returned transaction without signature")
+	} else if json.tx.Hash() != hash {
+		return nil, false, errors.New("invalid transaction hash")
 	}
 	if json.From != nil && json.BlockHash != nil {
 		setSenderFromServer(json.tx, *json.From, *json.BlockHash)
@@ -254,6 +284,10 @@ func (t *txAndReceipts) cacheBlockTxPositions(block *types.Block) {
 }
 
 func (t *txAndReceipts) sendTransaction(ctx context.Context, tx *types.Transaction) error {
+	sender, err := types.Sender(t.signer, tx)
+	if err != nil {
+		return nil
+	}
 	data, err := tx.MarshalBinary()
 	if err != nil {
 		return err
@@ -261,8 +295,93 @@ func (t *txAndReceipts) sendTransaction(ctx context.Context, tx *types.Transacti
 	if err := t.client.CallContext(ctx, nil, "eth_sendRawTransaction", hexutil.Encode(data)); err != nil {
 		return err
 	}
-	/*t.sentTxLock.Lock()
+	t.sentTxLock.Lock()
-	t.sentTxs[tx.]
+	t.markTxAsSeen(sender, tx)
-	t.sentTxLock.Unlock()*/
+	t.sentTxLock.Unlock()
 	return nil
 }
+
+func (t *txAndReceipts) markTxAsSeen(sender common.Address, tx *types.Transaction) {
+	senderTxs := t.sentTxs[sender]
+	if senderTxs == nil {
+		senderTxs = make(map[common.Hash]sentTx)
+		t.sentTxs[sender] = senderTxs
+	}
+	senderTxs[tx.Hash()] = sentTx{nonce: tx.Nonce(), lastSeen: t.headCounter}
+}
+
+func (t *txAndReceipts) newHead(number uint64, hash common.Hash) {
+	t.sentTxLock.Lock()
+	if number > t.lastHeadNumber {
+		t.lastHeadNumber = number
+		t.headCounter++
+		t.discardOldTxs()
+	}
+	t.sentTxLock.Unlock()
+}
+
+func (t *txAndReceipts) discardOldTxs() {
+	for sender, senderTxs := range t.sentTxs {
+		for txHash, sentTx := range senderTxs {
+			if sentTx.lastSeen+maxTxAge < t.headCounter {
+				delete(senderTxs, txHash)
+			}
+		}
+		if len(senderTxs) == 0 {
+			delete(t.sentTxs, sender)
+		}
+	}
+}
+
+func (t *txAndReceipts) nonceAndPendingTxs(ctx context.Context, head *btypes.ExecutionHeader, sender common.Address) (uint64, types.Transactions, error) {
+	proof, err := t.lightState.fetchProof(ctx, proofRequest{blockNumber: head.BlockNumber(), address: sender, storageKeys: ""})
+	if err != nil {
+		return 0, nil, err
+	}
+	if err := t.lightState.validateProof(proof, head.StateRoot(), sender, nil); err != nil {
+		return 0, nil, err
+	}
+
+	t.sentTxLock.Lock()
+	senderTxs := t.sentTxs[sender]
+	t.sentTxLock.Unlock()
+
+	type pendingResult struct {
+		pendingTx *types.Transaction
+		err       error
+	}
+
+	resultCh := make(chan pendingResult, len(senderTxs))
+	var reqCount int
+	for txHash, sentTx := range senderTxs {
+		if sentTx.nonce <= proof.Nonce {
+			continue
+		}
+		reqCount++
+		go func() {
+			tx, isPending, err := t.getUncachedTxByHash(ctx, txHash, head.BlockNumber())
+			if err == nil && isPending {
+				resultCh <- pendingResult{pendingTx: tx}
+			} else {
+				resultCh <- pendingResult{err: err}
+			}
+		}()
+	}
+	pendingList := make(types.Transactions, 0, len(senderTxs))
+	for ; reqCount > 0; reqCount-- {
+		res := <-resultCh
+		if res.err != nil {
+			return 0, nil, err
+		}
+		if res.pendingTx != nil {
+			pendingList = append(pendingList, res.pendingTx)
+		}
+	}
+	sort.Sort(types.TxByNonce(pendingList))
+	t.sentTxLock.Lock()
+	for _, tx := range pendingList {
+		t.markTxAsSeen(sender, tx)
+	}
+	t.sentTxLock.Unlock()
+	return proof.Nonce, pendingList, nil
+}