Another fix for body download (erigontech#6633)

When body gets evicted from the cache, its corresponding entry in
`requestMap` is also removed but was never re-instated when re-request
happens

Co-authored-by: Alexey Sharp <alexeysharp@Alexeys-iMac.local>

eth/stagedsync/stage_bodies.go

@@ -200,7 +200,7 @@ func BodiesForward(
 		write := true
 		for i := uint64(0); i < toProcess; i++ {
 			nextBlock := requestedLow + i
-			rawBody := cfg.bd.GetBodyFromCache(nextBlock)
+			rawBody := cfg.bd.GetBodyFromCache(nextBlock, write /* delete */)
 			if rawBody == nil {
 				log.Debug("Body was nil when reading from cache", "block", nextBlock)
 				cfg.bd.NotDelivered(nextBlock)
@@ -209,6 +209,7 @@ func BodiesForward(
 			if !write {
 				continue
 			}
+			cfg.bd.NotDelivered(nextBlock)
 			header, _, err := cfg.bd.GetHeader(nextBlock, cfg.blockReader, tx)
 			if err != nil {
 				return false, err

turbo/stages/bodydownload/body_algos.go

@@ -137,17 +137,7 @@ func (bd *BodyDownload) RequestMoreBodies(tx kv.RwTx, blockReader services.FullB
 					(header.WithdrawalsHash != nil && *header.WithdrawalsHash != types.EmptyRootHash) {
 					// Perhaps we already have this block
 					block := rawdb.ReadBlock(tx, hash, blockNum)
-					if block == nil {
-						var tripleHash TripleHash
-						copy(tripleHash[:], header.UncleHash.Bytes())
-						copy(tripleHash[length.Hash:], header.TxHash.Bytes())
-						if header.WithdrawalsHash != nil {
-							copy(tripleHash[2*length.Hash:], header.WithdrawalsHash.Bytes())
-						} else {
-							copy(tripleHash[2*length.Hash:], types.EmptyRootHash.Bytes())
-						}
-						bd.requestedMap[tripleHash] = blockNum
-					} else {
+					if block != nil {
 						bd.addBodyToCache(blockNum, block.RawBody())
 						request = false
 					}
@@ -159,6 +149,15 @@ func (bd *BodyDownload) RequestMoreBodies(tx kv.RwTx, blockReader services.FullB
 		}
 
 		if request {
+			var tripleHash TripleHash
+			copy(tripleHash[:], header.UncleHash.Bytes())
+			copy(tripleHash[length.Hash:], header.TxHash.Bytes())
+			if header.WithdrawalsHash != nil {
+				copy(tripleHash[2*length.Hash:], header.WithdrawalsHash.Bytes())
+			} else {
+				copy(tripleHash[2*length.Hash:], types.EmptyRootHash.Bytes())
+			}
+			bd.requestedMap[tripleHash] = blockNum
 			blockNums = append(blockNums, blockNum)
 			hashes = append(hashes, hash)
 		} else {
@@ -327,8 +326,7 @@ Loop:
 // the requestedLow count is increased by the number returned
 func (bd *BodyDownload) NextProcessingCount() uint64 {
 	var i uint64
-	for i = 0; !bd.delivered.IsEmpty() && bd.requestedLow+i == bd.delivered.Minimum(); i++ {
-		bd.delivered.Remove(bd.requestedLow + i)
+	for i = 0; bd.delivered.Contains(bd.requestedLow + i); i++ {
 	}
 	return i
 }
@@ -395,7 +393,7 @@ func (bd *BodyDownload) GetHeader(blockNum uint64, blockReader services.FullBloc
 func (bd *BodyDownload) addBodyToCache(key uint64, body *types.RawBody) {
 	size := body.EncodingSize()
 	if item, ok := bd.bodyCache.Get(BodyTreeItem{blockNum: key}); ok {
-		bd.bodyCacheSize -= item.payloadSize // It will be replace, so subtracting
+		bd.bodyCacheSize -= item.payloadSize // It will be replaced, so subtracting
 	}
 	bd.bodyCache.ReplaceOrInsert(BodyTreeItem{payloadSize: size, blockNum: key, rawBody: body})
 	bd.bodyCacheSize += size
@@ -406,10 +404,16 @@ func (bd *BodyDownload) addBodyToCache(key uint64, body *types.RawBody) {
 	}
 }
 
-func (bd *BodyDownload) GetBodyFromCache(blockNum uint64) *types.RawBody {
-	if item, ok := bd.bodyCache.Delete(BodyTreeItem{blockNum: blockNum}); ok {
-		bd.bodyCacheSize -= item.payloadSize
-		return item.rawBody
+func (bd *BodyDownload) GetBodyFromCache(blockNum uint64, delete bool) *types.RawBody {
+	if delete {
+		if item, ok := bd.bodyCache.Delete(BodyTreeItem{blockNum: blockNum}); ok {
+			bd.bodyCacheSize -= item.payloadSize
+			return item.rawBody
+		}
+	} else {
+		if item, ok := bd.bodyCache.Get(BodyTreeItem{blockNum: blockNum}); ok {
+			return item.rawBody
+		}
 	}
 	return nil
 }