fast-path for getReceipts (#12050)

erigontech · Sep 29, 2024 · ad05810 · ad05810
1 parent 443fabe
commit ad05810
Show file tree

Hide file tree

Showing 5 changed files with 102 additions and 32 deletions.
diff --git a/eth/protocols/eth/handlers.go b/eth/protocols/eth/handlers.go
@@ -162,16 +162,67 @@ func AnswerGetBlockBodiesQuery(db kv.Tx, query GetBlockBodiesPacket, blockReader
 
 type ReceiptsGetter interface {
 	GetReceipts(ctx context.Context, cfg *chain.Config, tx kv.Tx, block *types.Block) (types.Receipts, error)
+	GetCachedReceipts(ctx context.Context, blockHash libcommon.Hash) (types.Receipts, bool)
 }
 
-func AnswerGetReceiptsQuery(ctx context.Context, cfg *chain.Config, receiptsGetter ReceiptsGetter, br services.FullBlockReader, db kv.Tx, query GetReceiptsPacket) ([]rlp.RawValue, error) { //nolint:unparam
-	// Gather state data until the fetch or network limits is reached
+type cachedReceipts struct {
+	EncodedReceipts []rlp.RawValue
+	Bytes           int // total size of the encoded receipts
+	PendingIndex    int // index of the first not-found receipt in the query
+}
+
+func AnswerGetReceiptsQueryCacheOnly(ctx context.Context, receiptsGetter ReceiptsGetter, query GetReceiptsPacket) (*cachedReceipts, bool, error) {
 	var (
-		bytes    int
-		receipts []rlp.RawValue
+		bytes        int
+		receiptsList []rlp.RawValue
+		pendingIndex int
+		needMore     = true
 	)
 
 	for lookups, hash := range query {
+		if bytes >= softResponseLimit || len(receiptsList) >= maxReceiptsServe ||
+			lookups >= 2*maxReceiptsServe {
+			needMore = false
+			break
+		}
+		if receipts, ok := receiptsGetter.GetCachedReceipts(ctx, hash); ok {
+			if encoded, err := rlp.EncodeToBytes(receipts); err != nil {
+				return nil, needMore, fmt.Errorf("failed to encode receipt: %w", err)
+			} else {
+				receiptsList = append(receiptsList, encoded)
+				bytes += len(encoded)
+				pendingIndex = lookups + 1
+			}
+		} else {
+			break
+		}
+	}
+	if pendingIndex == len(query) {
+		needMore = false
+	}
+	return &cachedReceipts{
+		EncodedReceipts: receiptsList,
+		Bytes:           bytes,
+		PendingIndex:    pendingIndex,
+	}, needMore, nil
+}
+
+func AnswerGetReceiptsQuery(ctx context.Context, cfg *chain.Config, receiptsGetter ReceiptsGetter, br services.FullBlockReader, db kv.Tx, query GetReceiptsPacket, cachedReceipts *cachedReceipts) ([]rlp.RawValue, error) { //nolint:unparam
+	// Gather state data until the fetch or network limits is reached
+	var (
+		bytes        int
+		receipts     []rlp.RawValue
+		pendingIndex int
+	)
+
+	if cachedReceipts != nil {
+		bytes = cachedReceipts.Bytes
+		receipts = cachedReceipts.EncodedReceipts
+		pendingIndex = cachedReceipts.PendingIndex
+	}
+
+	for lookups := pendingIndex; lookups < len(query); lookups++ {
+		hash := query[lookups]
 		if bytes >= softResponseLimit || len(receipts) >= maxReceiptsServe ||
 			lookups >= 2*maxReceiptsServe {
 			break

diff --git a/p2p/sentry/sentry_multi_client/sentry_multi_client.go b/p2p/sentry/sentry_multi_client/sentry_multi_client.go
@@ -576,26 +576,35 @@ func (cs *MultiClient) getReceipts66(ctx context.Context, inreq *proto_sentry.In
 	if !EnableP2PReceipts {
 		return nil
 	}
-
-	err := cs.getReceiptsActiveGoroutineNumber.Acquire(ctx, 1)
-	if err != nil {
-		return err
-	}
-	defer cs.getReceiptsActiveGoroutineNumber.Release(1)
 	var query eth.GetReceiptsPacket66
 	if err := rlp.DecodeBytes(inreq.Data, &query); err != nil {
 		return fmt.Errorf("decoding getReceipts66: %w, data: %x", err, inreq.Data)
 	}
-
-	tx, err := cs.db.BeginRo(ctx)
+	cachedReceipts, needMore, err := eth.AnswerGetReceiptsQueryCacheOnly(ctx, cs.ethApiWrapper, query.GetReceiptsPacket)
 	if err != nil {
 		return err
 	}
-	defer tx.Rollback()
+	receiptsList := []rlp.RawValue{}
+	if cachedReceipts != nil {
+		receiptsList = cachedReceipts.EncodedReceipts
+	}
+	if needMore {
+		err = cs.getReceiptsActiveGoroutineNumber.Acquire(ctx, 1)
+		if err != nil {
+			return err
+		}
+		defer cs.getReceiptsActiveGoroutineNumber.Release(1)
+
+		tx, err := cs.db.BeginRo(ctx)
+		if err != nil {
+			return err
+		}
+		defer tx.Rollback()
+		receiptsList, err = eth.AnswerGetReceiptsQuery(ctx, cs.ChainConfig, cs.ethApiWrapper, cs.blockReader, tx, query.GetReceiptsPacket, cachedReceipts)
+		if err != nil {
+			return err
+		}
 
-	receiptsList, err := eth.AnswerGetReceiptsQuery(ctx, cs.ChainConfig, cs.ethApiWrapper, cs.blockReader, tx, query.GetReceiptsPacket)
-	if err != nil {
-		return err
 	}
 	b, err := rlp.EncodeToBytes(&eth.ReceiptsRLPPacket66{
 		RequestId:         query.RequestId,

diff --git a/turbo/jsonrpc/erigon_receipts.go b/turbo/jsonrpc/erigon_receipts.go
@@ -41,24 +41,26 @@ import (
 
 // GetLogsByHash implements erigon_getLogsByHash. Returns an array of arrays of logs generated by the transactions in the block given by the block's hash.
 func (api *ErigonImpl) GetLogsByHash(ctx context.Context, hash common.Hash) ([][]*types.Log, error) {
-	tx, err := api.db.BeginRo(ctx)
-	if err != nil {
-		return nil, err
-	}
-	defer tx.Rollback()
+	receipts, ok := api.getCachedReceipts(ctx, hash)
+	if !ok {
+		tx, err := api.db.BeginRo(ctx)
+		if err != nil {
+			return nil, err
+		}
+		defer tx.Rollback()
 
-	block, err := api.blockByHashWithSenders(ctx, tx, hash)
-	if err != nil {
-		return nil, err
-	}
-	if block == nil {
-		return nil, nil
-	}
-	receipts, err := api.getReceipts(ctx, tx, block)
-	if err != nil {
-		return nil, fmt.Errorf("getReceipts error: %w", err)
+		block, err := api.blockByHashWithSenders(ctx, tx, hash)
+		if err != nil {
+			return nil, err
+		}
+		if block == nil {
+			return nil, nil
+		}
+		receipts, err = api.getReceipts(ctx, tx, block)
+		if err != nil {
+			return nil, fmt.Errorf("getReceipts error: %w", err)
+		}
 	}
-
 	logs := make([][]*types.Log, len(receipts))
 	for i, receipt := range receipts {
 		logs[i] = receipt.Logs

diff --git a/turbo/jsonrpc/eth_receipts.go b/turbo/jsonrpc/eth_receipts.go
@@ -52,6 +52,10 @@ func (api *BaseAPI) getReceipts(ctx context.Context, tx kv.Tx, block *types.Bloc
 	return api.receiptsGenerator.GetReceipts(ctx, chainConfig, tx, block)
 }
 
+func (api *BaseAPI) getCachedReceipts(ctx context.Context, hash common.Hash) (types.Receipts, bool) {
+	return api.receiptsGenerator.GetCachedReceipts(ctx, hash)
+}
+
 // GetLogs implements eth_getLogs. Returns an array of logs matching a given filter object.
 func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (types.Logs, error) {
 	var begin, end uint64

diff --git a/turbo/jsonrpc/receipts/receipts_generator.go b/turbo/jsonrpc/receipts/receipts_generator.go
@@ -40,6 +40,10 @@ func NewGenerator(cacheSize int, blockReader services.FullBlockReader,
 	}
 }
 
+func (g *Generator) GetCachedReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, bool) {
+	return g.receiptsCache.Get(blockHash)
+}
+
 func (g *Generator) GetReceipts(ctx context.Context, cfg *chain.Config, tx kv.Tx, block *types.Block) (types.Receipts, error) {
 	if receipts, ok := g.receiptsCache.Get(block.Hash()); ok {
 		return receipts, nil