common, db, cl, execution, rpc, node: consolidate pool and retry utilities

cl/antiquary/antiquary_test.go

@@ -266,7 +266,7 @@ func TestAntiquateBytesListDiff(t *testing.T) {
 		return err
 	}
 
-	require.NoError(t, antiquateBytesListDiff(context.Background(), key, old, newData, collector, simpleDiff))
+	require.NoError(t, antiquateBytesListDiff(context.Background(), key, old, newData, &bytes.Buffer{}, collector, simpleDiff))
 
 	collected := collectAll(t, collector)
 	result, ok := collected[string(key)]
@@ -291,7 +291,7 @@ func TestAntiquateBytesListDiff_WithRealDiffFn(t *testing.T) {
 
 	key := base_encoding.Encode64ToBytes4(99)
 	require.NoError(t, antiquateBytesListDiff(
-		context.Background(), key, old, newData, collector,
+		context.Background(), key, old, newData, &bytes.Buffer{}, collector,
 		base_encoding.ComputeCompressedSerializedUint64ListDiff,
 	))
 

cl/antiquary/beacon_states_collector.go

@@ -519,11 +519,11 @@ func (i *beaconStatesCollector) collectStateEvents(slot uint64, events *state_ac
 }
 
 func (i *beaconStatesCollector) collectBalancesDiffs(ctx context.Context, slot uint64, old, new []byte) error {
-	return antiquateBytesListDiff(ctx, base_encoding.Encode64ToBytes4(slot), old, new, i.balancesCollector, base_encoding.ComputeCompressedSerializedUint64ListDiff)
+	return antiquateBytesListDiff(ctx, base_encoding.Encode64ToBytes4(slot), old, new, i.buf, i.balancesCollector, base_encoding.ComputeCompressedSerializedUint64ListDiff)
 }
 
 func (i *beaconStatesCollector) collectEffectiveBalancesDiffs(ctx context.Context, slot uint64, oldValidatorSetSSZ, newValidatorSetSSZ []byte) error {
-	return antiquateBytesListDiff(ctx, base_encoding.Encode64ToBytes4(slot), oldValidatorSetSSZ, newValidatorSetSSZ, i.effectiveBalanceCollector, base_encoding.ComputeCompressedSerializedEffectiveBalancesDiff)
+	return antiquateBytesListDiff(ctx, base_encoding.Encode64ToBytes4(slot), oldValidatorSetSSZ, newValidatorSetSSZ, i.buf, i.effectiveBalanceCollector, base_encoding.ComputeCompressedSerializedEffectiveBalancesDiff)
 }
 
 func (i *beaconStatesCollector) collectInactivityScores(slot uint64, inactivityScores []byte) error {
@@ -717,15 +717,13 @@ func antiquateListSSZ[T solid.EncodableHashableSSZ](ctx context.Context, slot ui
 	return collector.Collect(base_encoding.Encode64ToBytes4(roundedSlot), buffer.Bytes())
 }
 
-func antiquateBytesListDiff(ctx context.Context, key []byte, old, new []byte, collector *etl.Collector, diffFn func(w io.Writer, old, new []byte) error) error {
-	// create a diff
-	diffBuffer := bufferPool.Get().(*bytes.Buffer)
-	defer bufferPool.Put(diffBuffer)
-	diffBuffer.Reset()
+func antiquateBytesListDiff(ctx context.Context, key []byte, old, new []byte, buffer *bytes.Buffer, collector *etl.Collector, diffFn func(w io.Writer, old, new []byte) error) error {
+	buffer.Reset()
 
-	if err := diffFn(diffBuffer, old, new); err != nil {
+	// create a diff
+	if err := diffFn(buffer, old, new); err != nil {
 		return err
 	}
 
-	return collector.Collect(key, diffBuffer.Bytes())
+	return collector.Collect(key, buffer.Bytes())
 }

cl/antiquary/state_antiquary.go

@@ -43,13 +43,6 @@ import (
 	"github.com/erigontech/erigon/db/snaptype"
 )
 
-// pool for buffers
-var bufferPool = sync.Pool{
-	New: func() any {
-		return &bytes.Buffer{}
-	},
-}
-
 func (s *Antiquary) loopStates(ctx context.Context) {
 	// Execute this each second
 	reqRetryTimer := time.NewTicker(100 * time.Millisecond)

cl/persistence/base_encoding/uint64_diff.go

@@ -43,26 +43,13 @@ func putComp(v *zstd.Encoder) {
 	compressorPool.Put(v)
 }
 
-var bufferPool = sync.Pool{
-	New: func() any {
-		return &bytes.Buffer{}
-	},
-}
-
 var plainUint64BufferPool = sync.Pool{
 	New: func() any {
 		b := make([]uint64, 1028)
 		return &b
 	},
 }
 
-var plainBytesBufferPool = sync.Pool{
-	New: func() any {
-		b := make([]byte, 1028)
-		return &b
-	},
-}
-
 var repeatedPatternBufferPool = sync.Pool{
 	New: func() any {
 		b := make([]repeatedPatternEntry, 1028)
@@ -208,21 +195,15 @@ func ComputeCompressedSerializedEffectiveBalancesDiff(w io.Writer, old, new []by
 func ApplyCompressedSerializedUint64ListDiff(in, out []byte, diff []byte, reverse bool) ([]byte, error) {
 	out = out[:0]
 
-	buffer := bufferPool.Get().(*bytes.Buffer)
-	defer bufferPool.Put(buffer)
-	buffer.Reset()
-
-	if _, err := buffer.Write(diff); err != nil {
-		return nil, err
-	}
+	reader := bytes.NewReader(diff)
 
 	var length uint32
-	if err := binary.Read(buffer, binary.BigEndian, &length); err != nil {
+	if err := binary.Read(reader, binary.BigEndian, &length); err != nil {
 		return nil, err
 	}
 	var entry repeatedPatternEntry
 
-	decompressor, err := zstd.NewReader(buffer)
+	decompressor, err := zstd.NewReader(reader)
 	if err != nil {
 		return nil, err
 	}
@@ -307,19 +288,13 @@ func ApplyCompressedSerializedValidatorListDiff(in, out []byte, diff []byte, rev
 	}
 	out = out[:len(in)]
 
-	buffer := bufferPool.Get().(*bytes.Buffer)
-	defer bufferPool.Put(buffer)
-	buffer.Reset()
-
-	if _, err := buffer.Write(diff); err != nil {
-		return nil, err
-	}
+	reader := bytes.NewReader(diff)
 
 	currValidator := make([]byte, 121)
 
 	for {
 		var index uint32
-		if err := binary.Read(buffer, binary.BigEndian, &index); err != nil {
+		if err := binary.Read(reader, binary.BigEndian, &index); err != nil {
 			if errors.Is(err, io.EOF) {
 				break
 			}
@@ -328,7 +303,7 @@ func ApplyCompressedSerializedValidatorListDiff(in, out []byte, diff []byte, rev
 		if index == 1<<31 {
 			break
 		}
-		n, err := io.ReadFull(buffer, currValidator)
+		n, err := io.ReadFull(reader, currValidator)
 		if err != nil && !errors.Is(err, io.EOF) {
 			return nil, err
 		}
@@ -342,7 +317,7 @@ func ApplyCompressedSerializedValidatorListDiff(in, out []byte, diff []byte, rev
 		copy(out[index*121:], currValidator)
 	}
 	for {
-		n, err := io.ReadFull(buffer, currValidator)
+		n, err := io.ReadFull(reader, currValidator)
 		if err != nil && !errors.Is(err, io.EOF) {
 			return nil, err
 		}

cl/persistence/beacon_indicies/indicies.go

@@ -29,17 +29,11 @@ import (
 	"github.com/erigontech/erigon/cl/persistence/base_encoding"
 	"github.com/erigontech/erigon/cl/persistence/format/snapshot_format"
 	"github.com/erigontech/erigon/common"
+	"github.com/erigontech/erigon/common/pool"
 	"github.com/erigontech/erigon/db/kv"
 	"github.com/erigontech/erigon/db/kv/dbutils"
 )
 
-// make a buffer pool
-var bufferPool = &sync.Pool{
-	New: func() any {
-		return new(bytes.Buffer)
-	},
-}
-
 // make a zstd writer pool
 var zstdWriterPool = &sync.Pool{
 	New: func() any {
@@ -350,11 +344,10 @@ func WriteBeaconBlock(ctx context.Context, tx kv.RwTx, block *cltypes.SignedBeac
 		return err
 	}
 	// take a buffer and encoder
-	buf := bufferPool.Get().(*bytes.Buffer)
-	defer bufferPool.Put(buf)
+	buf := pool.GetBuffer()
+	defer pool.PutBuffer(buf)
 	encoder := zstdWriterPool.Get().(*zstd.Encoder)
 	defer putWriter(encoder)
-	buf.Reset()
 	encoder.Reset(buf)
 	_, err = snapshot_format.WriteBlockForSnapshot(encoder, block, nil)
 	if err != nil {

cl/persistence/format/snapshot_format/blocks.go

@@ -17,15 +17,14 @@
 package snapshot_format
 
 import (
-	"bytes"
 	"encoding/binary"
 	"io"
-	"sync"
 
 	"github.com/erigontech/erigon/cl/clparams"
 	"github.com/erigontech/erigon/cl/cltypes"
 	"github.com/erigontech/erigon/cl/cltypes/solid"
 	"github.com/erigontech/erigon/common"
+	"github.com/erigontech/erigon/common/pool"
 	"github.com/erigontech/erigon/execution/types"
 )
 
@@ -38,10 +37,6 @@ type ExecutionBlockReaderByNumber interface {
 	CacheBody(blockNumber uint64, transactions [][]byte, withdrawals []*types.Withdrawal)
 }
 
-var buffersPool = sync.Pool{
-	New: func() any { return &bytes.Buffer{} },
-}
-
 // WriteBlockForSnapshot writes a block to the given writer in the format expected by the snapshot.
 // buf is just a reusable buffer. if it had to grow it will be returned back as grown.
 //
@@ -103,9 +98,8 @@ func readMetadataForBlock(r io.Reader, b []byte) (clparams.StateVersion, common.
 }
 
 func ReadBlockFromSnapshot(r io.Reader, executionReader ExecutionBlockReaderByNumber, cfg *clparams.BeaconChainConfig) (*cltypes.SignedBeaconBlock, error) {
-	buffer := buffersPool.Get().(*bytes.Buffer)
-	defer buffersPool.Put(buffer)
-	buffer.Reset()
+	buffer := pool.GetBuffer()
+	defer pool.PutBuffer(buffer)
 
 	// Read the metadata
 	metadataSlab := make([]byte, 33)
@@ -160,9 +154,8 @@ func ReadBlockFromSnapshot(r io.Reader, executionReader ExecutionBlockReaderByNu
 
 // ReadBlockHeaderFromSnapshotWithExecutionData reads the beacon block header and the EL block number and block hash.
 func ReadBlockHeaderFromSnapshotWithExecutionData(r io.Reader, cfg *clparams.BeaconChainConfig) (*cltypes.SignedBeaconBlockHeader, uint64, common.Hash, error) {
-	buffer := buffersPool.Get().(*bytes.Buffer)
-	defer buffersPool.Put(buffer)
-	buffer.Reset()
+	buffer := pool.GetBuffer()
+	defer pool.PutBuffer(buffer)
 
 	// Read the metadata
 	metadataSlab := make([]byte, 33)
@@ -232,9 +225,8 @@ func ReadBlockHeaderFromSnapshotWithExecutionData(r io.Reader, cfg *clparams.Bea
 //     Callers that need full EL data should use ReadBlockFromSnapshot instead.
 //   - GLOAS: stored as full SignedBeaconBlock (no payload to blind); decoded directly.
 func ReadBeaconBlockBodyFromSnapshot(r io.Reader, cfg *clparams.BeaconChainConfig) (*cltypes.SignedBeaconBlock, error) {
-	buffer := buffersPool.Get().(*bytes.Buffer)
-	defer buffersPool.Put(buffer)
-	buffer.Reset()
+	buffer := pool.GetBuffer()
+	defer pool.PutBuffer(buffer)
 
 	// Read the metadata
 	metadataSlab := make([]byte, 33)