state-sync handling in eth/69

cmd/integration/commands/stages.go

@@ -1397,6 +1397,7 @@ func newSync(ctx context.Context, db kv.TemporalRwDB, miningConfig *buildercfg.M
 		maxBlockBroadcastPeers,
 		false, /* disableBlockDownload */
 		false, /* enableWitProtocol */
+		nil,   /* bridgeReader */
 		logger,
 	)
 	if err != nil {

core/blockchain.go

@@ -209,6 +209,7 @@ func ExecuteBlockEphemerally(
 			logs = append(logs, receipt.Logs...)
 		}
 
+		// PIP-74 state-sync receipt handling.
 		stateSyncReceipt := &types.Receipt{}
 		if chainConfig.Consensus == chain.BorConsensus && len(blockLogs) > 0 {
 			slices.SortStableFunc(blockLogs, func(i, j *types.Log) int { return cmp.Compare(i.Index, j.Index) })

eth/backend.go

@@ -709,6 +709,14 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 		}
 	}
 
+	// Create bridge.Reader for P2P state sync receipts.
+	// When Erigon nodes serve receipts to peers via P2P, they need bridge.Reader
+	// to include state sync transaction receipts in the response.
+	var bridgeReaderForP2P *bridge.Reader
+	if bridgeStore != nil && chainConfig.Bor != nil {
+		bridgeReaderForP2P = bridge.NewReader(bridgeStore, logger, chainConfig.Bor.StateReceiverContractAddress())
+	}
+
 	sentryMcDisableBlockDownload := chainConfig.Bor != nil || config.ElBlockDownloaderV2
 	backend.sentriesClient, err = sentry_multi_client.NewMultiClient(
 		backend.chainDB,
@@ -723,6 +731,7 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
 		maxBlockBroadcastPeers,
 		sentryMcDisableBlockDownload,
 		stack.Config().P2P.EnableWitProtocol,
+		bridgeReaderForP2P,
 		logger,
 	)
 	if err != nil {

execution/stages/blockchain_test.go

@@ -307,7 +307,9 @@ func testReorgLong(t *testing.T) {
 
 // Tests that reorganising a short difficult chain after a long easy one
 // overwrites the canonical numbers and links in the database.
-func TestReorgShortBlocks(t *testing.T) { testReorgShort(t) }
+func TestReorgShortBlocks(t *testing.T) {
+	testReorgShort(t)
+}
 
 func testReorgShort(t *testing.T) {
 	t.Parallel()

execution/stages/mock/mock_sentry.go

@@ -323,7 +323,7 @@ func MockWithEverything(tb testing.TB, gspec *types.Genesis, key *ecdsa.PrivateK
 		PeerId:         gointerfaces.ConvertHashToH512([64]byte{0x12, 0x34, 0x50}), // "12345"
 		BlockSnapshots: allSnapshots,
 		BlockReader:    br,
-		ReceiptsReader: receipts.NewGenerator(br, engine, 5*time.Second),
+		ReceiptsReader: receipts.NewGenerator(br, engine, 5*time.Second, nil),
 		HistoryV3:      true,
 		cfg:            cfg,
 	}
@@ -441,6 +441,7 @@ func MockWithEverything(tb testing.TB, gspec *types.Genesis, key *ecdsa.PrivateK
 		maxBlockBroadcastPeers,
 		false, /* disableBlockDownload */
 		false, /* enableWitProtocol */
+		nil,   /* bridgeReader */
 		logger,
 	)
 	if err != nil {

execution/types/receipt.go

@@ -575,9 +575,25 @@ func (e receiptEncoder69) EncodeRLP(w io.Writer) error { return e.r.EncodeRLP69(
 
 func (rs Receipts) EncodeRLP69(w io.Writer) error {
 	encs := make([]receiptEncoder69, len(rs))
+	n := len(rs)
+
 	for i := range rs {
-		encs[i] = receiptEncoder69{r: rs[i]}
+		// Copy the receipt reference.
+		r := rs[i]
+
+		// Only the last receipt can be a state-sync tx.
+		if i == n-1 {
+			// Match the ReadStateSyncReceiptByHash logic:
+			// It's a state-sync transaction if the cumulative gas is zero, or
+			// the equal cumulative gas is equal to the previous one (zero gas usage)
+			if r.CumulativeGasUsed == 0 || (n >= 2 && r.CumulativeGasUsed == rs[n-2].CumulativeGasUsed) {
+				r.Type = 0
+			}
+		}
+
+		encs[i] = receiptEncoder69{r: r}
 	}
+
 	return rlp.Encode(w, encs)
 }
 

p2p/protocols/eth/protocol.go

@@ -181,6 +181,7 @@ type StatusPacket struct {
 type StatusPacket69 struct {
 	ProtocolVersion           uint32
 	NetworkID                 uint64
+	TD                        *big.Int
 	Genesis                   common.Hash
 	ForkID                    forkid.ID
 	MinimumBlock, LatestBlock uint64

p2p/sentry/eth_handshake.go

@@ -172,10 +172,12 @@ func encodeStatusPacket(status *sentryproto.StatusData, version uint) eth.Status
 }
 
 func encodeStatusPacket69(status *sentryproto.StatusData, version uint) eth.StatusPacket69 {
+	ourTD := gointerfaces.ConvertH256ToUint256Int(status.TotalDifficulty)
 	genesisHash := gointerfaces.ConvertH256ToHash(status.ForkData.Genesis)
 	return eth.StatusPacket69{
 		ProtocolVersion: uint32(version),
 		NetworkID:       status.NetworkId,
+		TD:              ourTD.ToBig(),
 		Genesis:         genesisHash,
 		ForkID:          forkid.NewIDFromForks(status.ForkData.HeightForks, status.ForkData.TimeForks, genesisHash, status.MaxBlockHeight, status.MaxBlockTime),
 		MinimumBlock:    status.MinimumBlockHeight,

p2p/sentry/sentry_multi_client/sentry_multi_client.go

@@ -284,6 +284,12 @@ type MultiClient struct {
 
 var _ eth.ReceiptsGetter = new(receipts.Generator) // compile-time interface-check
 
+// bridgeReader interface for reading bridge events (state sync).
+type bridgeReader interface {
+	Events(ctx context.Context, blockHash common.Hash, blockNum uint64) ([]*types.Message, error)
+	EventTxnLookup(ctx context.Context, borTxHash common.Hash) (uint64, bool, error)
+}
+
 func NewMultiClient(
 	db kv.TemporalRoDB,
 	chainConfig *chain.Config,
@@ -297,6 +303,7 @@ func NewMultiClient(
 	maxBlockBroadcastPeers func(*types.Header) uint,
 	disableBlockDownload bool,
 	enableWitProtocol bool,
+	bridgeReader bridgeReader,
 	logger log.Logger,
 ) (*MultiClient, error) {
 	// header downloader
@@ -338,6 +345,13 @@ func NewMultiClient(
 		witnessBuffer = stagedsync.NewWitnessBuffer()
 	}
 
+	// Create BorGenerator for state sync receipts if bridgeReader is provided.
+	// This is required for Erigon to include state sync transaction receipts in P2P GetReceipts responses.
+	var borGenerator *receipts.BorGenerator
+	if bridgeReader != nil {
+		borGenerator = receipts.NewBorGenerator(blockReader, engine, bridgeReader)
+	}
+
 	cs := &MultiClient{
 		Hd:                                hd,
 		Bd:                                bd,
@@ -354,7 +368,7 @@ func NewMultiClient(
 		disableBlockDownload:              disableBlockDownload,
 		logger:                            logger,
 		getReceiptsActiveGoroutineNumber:  semaphore.NewWeighted(1),
-		ethApiWrapper:                     receipts.NewGenerator(blockReader, engine, 5*time.Minute),
+		ethApiWrapper:                     receipts.NewGenerator(blockReader, engine, 5*time.Minute, borGenerator),
 	}
 
 	return cs, nil

rpc/jsonrpc/eth_api.go

@@ -157,6 +157,8 @@ func NewBaseApi(f *rpchelper.Filters, stateCache kvcache.Cache, blockReader serv
 		panic(err)
 	}
 
+	borReceiptGenerator := receipts.NewBorGenerator(blockReader, engine, bridgeReader)
+
 	return &BaseAPI{
 		filters:             f,
 		stateCache:          stateCache,
@@ -166,8 +168,8 @@ func NewBaseApi(f *rpchelper.Filters, stateCache kvcache.Cache, blockReader serv
 		_txNumReader:        blockReader.TxnumReader(context.Background()),
 		evmCallTimeout:      evmCallTimeout,
 		_engine:             engine,
-		receiptsGenerator:   receipts.NewGenerator(blockReader, engine, evmCallTimeout),
-		borReceiptGenerator: receipts.NewBorGenerator(blockReader, engine),
+		receiptsGenerator:   receipts.NewGenerator(blockReader, engine, evmCallTimeout, borReceiptGenerator),
+		borReceiptGenerator: borReceiptGenerator,
 		dirs:                dirs,
 		bridgeReader:        bridgeReader,
 	}

rpc/jsonrpc/eth_receipts.go

@@ -527,13 +527,27 @@ func (api *APIImpl) GetBlockReceipts(ctx context.Context, numberOrHash rpc.Block
 	if err != nil {
 		return nil, fmt.Errorf("getReceipts error: %w", err)
 	}
+
+	numTxs := len(block.Transactions())
 	result := make([]map[string]interface{}, 0, len(receipts))
+
 	for _, receipt := range receipts {
-		txn := block.Transactions()[receipt.TransactionIndex]
-		result = append(result, ethutils.MarshalReceipt(receipt, txn, chainConfig, block.HeaderNoCopy(), txn.Hash(), true, true))
+		// State-sync receipts' TransactionIndex is equal to numTxs.
+		if int(receipt.TransactionIndex) == numTxs {
+			// This is a state-sync transaction receipt.
+			result = append(result, ethutils.MarshalReceipt(receipt, bortypes.NewBorTransaction(), chainConfig, block.HeaderNoCopy(), receipt.TxHash, false, true))
+		} else {
+			// This is a normal transaction receipt.
+			txn := block.Transactions()[receipt.TransactionIndex]
+			result = append(result, ethutils.MarshalReceipt(receipt, txn, chainConfig, block.HeaderNoCopy(), txn.Hash(), true, true))
+		}
 	}
 
-	if chainConfig.Bor != nil {
+	// TODO: Change Rio to the actual hard fork.
+	// Rio fork (or later) includes state-sync receipts in block execution.
+	// For blocks >= Rio: receipts already include the state-sync receipts.
+	// For blocks < Rio: old cached data doesn't include the state-sync receipts, generate on-the-fly.
+	if chainConfig.Bor != nil && !chainConfig.Bor.IsRio(blockNum) && len(receipts) == numTxs {
 		events, err := api.bridgeReader.Events(ctx, block.Hash(), blockNum)
 		if err != nil {
 			return nil, err

rpc/jsonrpc/receipts/bor_receipts_generator.go

@@ -21,14 +21,19 @@ import (
 	"github.com/erigontech/erigon/turbo/transactions"
 )
 
+type bridgeReader interface {
+	Events(ctx context.Context, blockHash common.Hash, blockNum uint64) ([]*types.Message, error)
+	EventTxnLookup(ctx context.Context, borTxHash common.Hash) (uint64, bool, error)
+}
+
 type BorGenerator struct {
 	receiptCache *lru.Cache[common.Hash, *types.Receipt]
 	blockReader  services.FullBlockReader
 	engine       consensus.EngineReader
+	bridgeReader bridgeReader
 }
 
-func NewBorGenerator(blockReader services.FullBlockReader,
-	engine consensus.EngineReader) *BorGenerator {
+func NewBorGenerator(blockReader services.FullBlockReader, engine consensus.EngineReader, bridgeReader bridgeReader) *BorGenerator {
 	receiptCache, err := lru.New[common.Hash, *types.Receipt](receiptsCacheLimit)
 	if err != nil {
 		panic(err)
@@ -38,6 +43,7 @@ func NewBorGenerator(blockReader services.FullBlockReader,
 		receiptCache: receiptCache,
 		blockReader:  blockReader,
 		engine:       engine,
+		bridgeReader: bridgeReader,
 	}
 }