eth, core: handle state-sync data in eth/69 messages

core/blockchain.go

@@ -1659,6 +1659,98 @@ const (
 	SideStatTy
 )
 
+// getReceiptFields given a list of normal receipts returns the tx index, the log index
+// and cumulative gas used for populating the bor receipt.
+func getReceiptFields(receipts []*types.ReceiptForStorage) (int, int, uint64) {
+	if len(receipts) == 0 {
+		return 0, 0, 0
+	}
+
+	logs := 0
+	for _, receipt := range receipts {
+		logs += len(receipt.Logs)
+	}
+
+	cumulativeGasUsed := receipts[len(receipts)-1].CumulativeGasUsed
+
+	return len(receipts), logs, cumulativeGasUsed
+}
+
+// isStateSyncReceiptPresent checks if a state-sync receipt is present in the list of
+// receipts or not.
+func isStateSyncReceiptPresent(decoded []*types.ReceiptForStorage) bool {
+	if len(decoded) == 0 {
+		return false
+	}
+
+	// The state-sync receipt can either have a 0 cumulative gas used (this depends on the remote peer) or
+	// have the same cumulative gas used as the previous receipt as state-sync transactions uses 0 gas and
+	// hence they don't contribute to the cumulative gas used value.
+	if decoded[len(decoded)-1].CumulativeGasUsed == 0 {
+		return true
+	}
+
+	if len(decoded) >= 2 && decoded[len(decoded)-1].CumulativeGasUsed == decoded[len(decoded)-2].CumulativeGasUsed {
+		return true
+	}
+
+	return false
+}
+
+// splitReceiptsAndDeriveFields separates out the state-sync receipt from the whole receipt list
+// of a block and returns the encoded lists back separately. If a state-sync receipt is found, it
+// derives the necessary fields and populates them. In case of errors or empty receipt, it returns
+// `nil` instead of `rlp.EncodeToBytes(nil)`.
+func splitReceiptsAndDeriveFields(receipts rlp.RawValue, number uint64, hash common.Hash, borCfg *params.BorConfig) (rlp.RawValue, rlp.RawValue) {
+	if receipts == nil {
+		return nil, nil
+	}
+
+	// Bor receipts can only exist on sprint end blocks. Avoid decoding if possible.
+	if !types.IsSprintEndBlock(borCfg, number) {
+		return receipts, nil
+	}
+
+	var decoded []*types.ReceiptForStorage
+	if err := rlp.DecodeBytes(receipts, &decoded); err != nil {
+		log.Warn("Failed to decode block receipts", "number", number, "hash", hash, "err", err)
+		return receipts, nil
+	}
+
+	// Split receipts only if there's a state-sync receipt present
+	if isStateSyncReceiptPresent(decoded) {
+		borReceipt := decoded[len(decoded)-1]
+
+		// Derive rest of fields for bor receipts before encoding back
+		txIndex, logIndex, cumulativeGasUsed := getReceiptFields(decoded[:len(decoded)-1])
+		types.DeriveFieldsForBorLogs(borReceipt.Logs, hash, number, uint(txIndex), uint(logIndex))
+		borReceipt.Status = types.ReceiptStatusSuccessful
+		borReceipt.CumulativeGasUsed = cumulativeGasUsed
+
+		// Encode the state-sync transaction receipt separately
+		encodedStateSyncReceipt, err := rlp.EncodeToBytes(borReceipt)
+		if err != nil {
+			log.Warn("Failed to encode state-sync receipt", "number", number, "hash", hash, "err", err)
+			return receipts, nil
+		}
+
+		// If no receipts left, return
+		if len(decoded[:len(decoded)-1]) == 0 {
+			return nil, encodedStateSyncReceipt
+		}
+
+		// Encode back the normal (non state-sync) receipts and return
+		encodedReceipts, err := rlp.EncodeToBytes(decoded[:len(decoded)-1])
+		if err != nil {
+			log.Warn("Failed to encode remaining receipts after excluding state-sync receipt", "number", number, "hash", hash, "err", err)
+			return receipts, encodedStateSyncReceipt
+		}
+		return encodedReceipts, encodedStateSyncReceipt
+	}
+
+	return receipts, nil
+}
+
 // InsertReceiptChain attempts to complete an already existing header chain with
 // transaction and receipt data.
 func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain []rlp.RawValue, ancientLimit uint64) (int, error) {
@@ -1763,14 +1855,15 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 				log.Info("Wrote genesis to ancients")
 			}
 		}
-		// BOR: Retrieve all the bor receipts and also maintain the array of headers
-		// for bor specific reorg check.
-		borReceipts := []rlp.RawValue{}
 
-		var headers []*types.Header
+		// Separate out bor receipts (i.e. receipts of state-sync transactions)
+		var borReceipts = make([]rlp.RawValue, len(receiptChain))
+		for i, receipts := range receiptChain {
+			receiptChain[i], borReceipts[i] = splitReceiptsAndDeriveFields(receipts, blockChain[i].NumberU64(), blockChain[i].Hash(), bc.chainConfig.Bor)
+		}
 
+		var headers []*types.Header
 		for _, block := range blockChain {
-			borReceipts = append(borReceipts, bc.GetBorReceiptRLPByHash(block.Hash()))
 			headers = append(headers, block.Header())
 		}
 
@@ -1798,8 +1891,14 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 		for i, block := range blockChain {
 			if bc.txIndexer == nil || bc.txIndexer.limit == 0 || ancientLimit <= bc.txIndexer.limit || block.NumberU64() >= ancientLimit-bc.txIndexer.limit {
 				rawdb.WriteTxLookupEntriesByBlock(batch, block)
+				if len(borReceipts[i]) > 0 {
+					rawdb.WriteBorTxLookupEntry(batch, block.Hash(), block.NumberU64())
+				}
 			} else if rawdb.ReadTxIndexTail(bc.db) != nil {
 				rawdb.WriteTxLookupEntriesByBlock(batch, block)
+				if len(borReceipts[i]) > 0 {
+					rawdb.WriteBorTxLookupEntry(batch, block.Hash(), block.NumberU64())
+				}
 			}
 
 			stats.processed++
@@ -1897,11 +1996,24 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 					skipPresenceCheck = true
 				}
 			}
+
+			// Separate out bor receipts (i.e. receipts of state-sync transactions)
+			var borReceiptRaw rlp.RawValue
+			receiptChain[i], borReceiptRaw = splitReceiptsAndDeriveFields(receiptChain[i], block.NumberU64(), block.Hash(), bc.chainConfig.Bor)
+
 			// Write all the data out into the database
 			rawdb.WriteCanonicalHash(batch, block.Hash(), block.NumberU64())
 			rawdb.WriteBlock(batch, block)
 			rawdb.WriteRawReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i])
 
+			var borReceipt types.ReceiptForStorage
+			if len(borReceiptRaw) > 0 {
+				if err := rlp.DecodeBytes(borReceiptRaw, &borReceipt); err == nil {
+					rawdb.WriteBorReceipt(batch, block.Hash(), block.NumberU64(), &borReceipt)
+					rawdb.WriteBorTxLookupEntry(batch, block.Hash(), block.NumberU64())
+				}
+			}
+
 			// Write everything belongs to the blocks into the database. So that
 			// we can ensure all components of body is completed(body, receipts)
 			// except transaction indexes(will be created once sync is finished).

core/blockchain_test.go

@@ -49,6 +49,7 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/holiman/uint256"
+	"github.com/stretchr/testify/require"
 )
 
 // So we can deterministically seed different blockchains
@@ -5476,3 +5477,105 @@ func TestStatelessSetHeadBeyondRoot(t *testing.T) {
 		}
 	}
 }
+
+// TestSplitReceiptsAndDeriveFields checks if normal and state-sync receipts are split correctly
+// before inserting to database. It also checks if state-sync receipt fields are derived correctly.
+func TestSplitReceiptsAndDeriveFields(t *testing.T) {
+	var tests = []struct {
+		name             string
+		normalReceipts   []*types.ReceiptForStorage
+		stateSyncReceipt *types.ReceiptForStorage
+	}{
+		// Both nil
+		{
+			name:             "both normal and state-sync receipt is nil",
+			normalReceipts:   nil,
+			stateSyncReceipt: nil,
+		},
+		// Normal receipt empty and state-sync nil
+		{
+			name:             "empty normal receipt and nil state-sync receipt",
+			normalReceipts:   []*types.ReceiptForStorage{},
+			stateSyncReceipt: nil,
+		},
+		// Only normal receipt (single)
+		{
+			name:             "state-sync receipt is nil, only single normal receipt present",
+			normalReceipts:   []*types.ReceiptForStorage{{CumulativeGasUsed: 555, Status: 1, Logs: nil}},
+			stateSyncReceipt: nil,
+		},
+		// Only normal receipt (multiple)
+		{
+			name:             "state-sync receipt is nil, multiple normal receipts present",
+			normalReceipts:   []*types.ReceiptForStorage{{CumulativeGasUsed: 555, Status: 1, Logs: nil}, {CumulativeGasUsed: 666, Status: 1, Logs: nil}},
+			stateSyncReceipt: nil,
+		},
+		// Only state-sync receipt
+		{
+			name:             "normal receipt is nil, only state-sync receipt present",
+			normalReceipts:   nil,
+			stateSyncReceipt: &types.ReceiptForStorage{CumulativeGasUsed: 0, Status: 1, Logs: nil, Type: 0},
+		},
+		// Normal + state-sync receipts (single)
+		{
+			name:             "single normal and state-sync receipt present",
+			normalReceipts:   []*types.ReceiptForStorage{{CumulativeGasUsed: 555, Status: 1, Logs: nil}},
+			stateSyncReceipt: &types.ReceiptForStorage{CumulativeGasUsed: 0, Status: 1, Logs: nil, Type: 0},
+		},
+		// Normal + state-sync receipts (multiple)
+		{
+			name:             "multiple normal and single state-sync receipt present",
+			normalReceipts:   []*types.ReceiptForStorage{{CumulativeGasUsed: 555, Status: 1, Logs: nil}, {CumulativeGasUsed: 666, Status: 1, Logs: nil}, {CumulativeGasUsed: 777, Status: 1, Logs: nil}},
+			stateSyncReceipt: &types.ReceiptForStorage{CumulativeGasUsed: 0, Status: 1, Logs: nil, Type: 0},
+		},
+		// Normal + state-sync receipts (multiple) with non-zero cumulative gas used
+		{
+			name:             "multiple normal and single state-sync receipt present with non-zero cumulative gas used",
+			normalReceipts:   []*types.ReceiptForStorage{{CumulativeGasUsed: 555, Status: 1, Logs: nil}, {CumulativeGasUsed: 666, Status: 1, Logs: nil}, {CumulativeGasUsed: 777, Status: 1, Logs: nil}},
+			stateSyncReceipt: &types.ReceiptForStorage{CumulativeGasUsed: 777, Status: 1, Logs: nil, Type: 0},
+		},
+	}
+
+	// Create a mock config with sprint length as 1 (so that split receipts will run for all test cases)
+	mockBorCfg := params.BorConfig{
+		Sprint: map[string]uint64{"0": 1},
+	}
+
+	for _, test := range tests {
+		// Individually encode receipts for comparing with values after splitting
+		var normalEncoded []byte = nil
+		if len(test.normalReceipts) > 0 {
+			normalEncoded, _ = rlp.EncodeToBytes(test.normalReceipts)
+		}
+
+		// For state-sync receipts, create a copy and populate remaining fields which will be
+		// used to compare later with the output. Don't populate the original test object as
+		// the `splitReceiptsAndDeriveFields` should populate values in it.
+		stateSyncReceipt := test.stateSyncReceipt
+		var stateSyncEncoded []byte = nil
+		if test.stateSyncReceipt != nil {
+			_, _, cumulativeGasUsed := getReceiptFields(test.normalReceipts)
+			stateSyncReceipt.CumulativeGasUsed = cumulativeGasUsed
+			stateSyncEncoded, _ = rlp.EncodeToBytes(stateSyncReceipt)
+		}
+
+		// Merge both and encode the final list. Use the receipts in test object.
+		var allReceipts = make([]*types.ReceiptForStorage, 0)
+		if test.normalReceipts != nil {
+			allReceipts = append(allReceipts, test.normalReceipts...)
+		}
+		if test.stateSyncReceipt != nil {
+			allReceipts = append(allReceipts, test.stateSyncReceipt)
+		}
+		encoded, _ := rlp.EncodeToBytes(allReceipts)
+		if len(allReceipts) == 0 {
+			// Skip encoding, instead just set nil
+			encoded = nil
+		}
+
+		// Split receipts and assert if the individual list match with the expected receipt data or not
+		normal, stateSync := splitReceiptsAndDeriveFields(encoded, 0, common.Hash{}, &mockBorCfg)
+		require.Equal(t, rlp.RawValue(normalEncoded), normal, fmt.Sprintf("case: %s, normal receipts mismatch, got: %v, expected: %v", test.name, normal, normalEncoded))
+		require.Equal(t, rlp.RawValue(stateSyncEncoded), stateSync, fmt.Sprintf("case: %s, state-sync receipts mismatch, got: %v, expected: %v", test.name, stateSync, stateSyncEncoded))
+	}
+}

core/bor_blockchain.go → core/bor_blockchain_reader.go

@@ -4,14 +4,24 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
+var (
+	borReceiptsCacheHit     = metrics.NewRegisteredGauge("bor/receipts/cache/hit", nil)
+	borReceiptsCacheMiss    = metrics.NewRegisteredGauge("bor/receipts/cache/miss", nil)
+	borReceiptsRLPCacheHit  = metrics.NewRegisteredGauge("bor/rlpreceipts/cache/hit", nil)
+	borReceiptsRLPCacheMiss = metrics.NewRegisteredGauge("bor/rlpreceipts/cache/miss", nil)
+)
+
 // GetBorReceiptByHash retrieves the bor block receipt in a given block.
 func (bc *BlockChain) GetBorReceiptByHash(hash common.Hash) *types.Receipt {
 	if receipt, ok := bc.borReceiptsCache.Get(hash); ok {
+		borReceiptsCacheHit.Update(1)
 		return receipt
 	}
+	borReceiptsCacheMiss.Update(1)
 
 	// read header from hash
 	number := rawdb.ReadHeaderNumber(bc.db, hash)
@@ -34,8 +44,10 @@ func (bc *BlockChain) GetBorReceiptByHash(hash common.Hash) *types.Receipt {
 // GetBorReceiptRLPByHash retrieves the bor block receipt RLP in a given block.
 func (bc *BlockChain) GetBorReceiptRLPByHash(hash common.Hash) rlp.RawValue {
 	if receiptRLP, ok := bc.borReceiptsRLPCache.Get(hash); ok {
+		borReceiptsRLPCacheHit.Update(1)
 		return receiptRLP
 	}
+	borReceiptsRLPCacheMiss.Update(1)
 
 	// read header from hash
 	number := rawdb.ReadHeaderNumber(bc.db, hash)

core/rawdb/accessors_history.go

@@ -160,20 +160,40 @@ func increaseKey(key []byte) []byte {
 //
 // Note, this method assumes the storage space with prefix `StateHistoryIndexPrefix`
 // is exclusively occupied by the history indexing data!
+//
+// Note, as bor receipts and tx lookup are also stored with prefix "matic-bor", it needs
+// to be excluded from range deletion.
 func DeleteStateHistoryIndex(db ethdb.KeyValueRangeDeleter) {
-	start := StateHistoryIndexPrefix
-	limit := increaseKey(bytes.Clone(StateHistoryIndexPrefix))
+	// We split the delete range queries into 2 parts to ensure that bor receipts related data isn't deleted.
+	// 1. `StateHistoryIndexPrefix` (i.e. "m") to "matic-bor" key (exclusive)
+	// 2. Immediate next key after "matic-bor" to next key after `StateHistoryIndexPrefix` (i.e. "n")
+	start1 := StateHistoryIndexPrefix
+	limit1 := []byte("matic-bor")
+	start2 := increaseKey(bytes.Clone(limit1))
+	limit2 := increaseKey(bytes.Clone(StateHistoryIndexPrefix))
 
 	// Try to remove the data in the range by a loop, as the leveldb
 	// doesn't support the native range deletion.
 	for {
-		err := db.DeleteRange(start, limit)
+		err := db.DeleteRange(start1, limit1)
+		if err == nil {
+			break
+		}
+		if errors.Is(err, ethdb.ErrTooManyKeys) {
+			continue
+		}
+		log.Crit("Failed to delete history index range 1", "err", err)
+	}
+
+	// Try removing for second range
+	for {
+		err := db.DeleteRange(start2, limit2)
 		if err == nil {
 			return
 		}
 		if errors.Is(err, ethdb.ErrTooManyKeys) {
 			continue
 		}
-		log.Crit("Failed to delete history index range", "err", err)
+		log.Crit("Failed to delete history index range 2", "err", err)
 	}
 }