diff --git a/cl/antiquary/antiquary_test.go b/cl/antiquary/antiquary_test.go
index 1a43fdb675a..3d6910f4cd0 100644
--- a/cl/antiquary/antiquary_test.go
+++ b/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,
))
diff --git a/cl/antiquary/beacon_states_collector.go b/cl/antiquary/beacon_states_collector.go
index 374032fb2a4..db3f8336639 100644
--- a/cl/antiquary/beacon_states_collector.go
+++ b/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())
}
diff --git a/cl/antiquary/state_antiquary.go b/cl/antiquary/state_antiquary.go
index f88b3ff27fe..ecad88dde2c 100644
--- a/cl/antiquary/state_antiquary.go
+++ b/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)
diff --git a/cl/persistence/base_encoding/uint64_diff.go b/cl/persistence/base_encoding/uint64_diff.go
index 9e07d8e35a8..53a5769d2a0 100644
--- a/cl/persistence/base_encoding/uint64_diff.go
+++ b/cl/persistence/base_encoding/uint64_diff.go
@@ -43,12 +43,6 @@ 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)
@@ -56,13 +50,6 @@ var plainUint64BufferPool = sync.Pool{
},
}
-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
}
diff --git a/cl/persistence/beacon_indicies/indicies.go b/cl/persistence/beacon_indicies/indicies.go
index 67aa6ed1411..2c59323b361 100644
--- a/cl/persistence/beacon_indicies/indicies.go
+++ b/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 {
diff --git a/cl/persistence/format/snapshot_format/blocks.go b/cl/persistence/format/snapshot_format/blocks.go
index d5ade0af4fb..66e3c839916 100644
--- a/cl/persistence/format/snapshot_format/blocks.go
+++ b/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)
diff --git a/cl/persistence/state/historical_states_reader/historical_states_reader.go b/cl/persistence/state/historical_states_reader/historical_states_reader.go
index abbadfd12d1..2b45cc7cd99 100644
--- a/cl/persistence/state/historical_states_reader/historical_states_reader.go
+++ b/cl/persistence/state/historical_states_reader/historical_states_reader.go
@@ -42,10 +42,6 @@ import (
"github.com/erigontech/erigon/db/snapshotsync/freezeblocks"
)
-var buffersPool = sync.Pool{
- New: func() any { return &bytes.Buffer{} },
-}
-
type HistoricalStatesReader struct {
cfg *clparams.BeaconChainConfig
validatorTable *state_accessors.StaticValidatorTable // We can save 80% of the I/O by caching the validator table
@@ -572,16 +568,8 @@ func (r *HistoricalStatesReader) reconstructDiffedUint64List(tx kv.Tx, kvGetter
return nil, fmt.Errorf("dump not found for slot %d", freshDumpSlot)
}
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
- buffer.Reset()
-
- if _, err := buffer.Write(compressed); err != nil {
- return nil, err
- }
-
// Read the diff file
- zstdReader, err := zstd.NewReader(buffer)
+ zstdReader, err := zstd.NewReader(bytes.NewReader(compressed))
if err != nil {
return nil, err
}
@@ -644,10 +632,6 @@ func (r *HistoricalStatesReader) reconstructBalances(tx kv.Tx, kvGetter state_ac
remainder := slot % clparams.SlotsPerDump
freshDumpSlot := slot - remainder
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
- buffer.Reset()
-
var compressed []byte
currentStageProgress, err := state_accessors.GetStateProcessingProgress(tx)
if err != nil {
@@ -670,10 +654,7 @@ func (r *HistoricalStatesReader) reconstructBalances(tx kv.Tx, kvGetter state_ac
if len(compressed) == 0 {
return nil, fmt.Errorf("dump not found for slot %d", freshDumpSlot)
}
- if _, err := buffer.Write(compressed); err != nil {
- return nil, err
- }
- zstdReader, err := zstd.NewReader(buffer)
+ zstdReader, err := zstd.NewReader(bytes.NewReader(compressed))
if err != nil {
return nil, err
}
@@ -1057,22 +1038,13 @@ func ReadQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors.GetValF
remainder := slot % clparams.SlotsPerDump
freshDumpSlot := slot - remainder
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
- buffer.Reset()
-
- var compressed []byte
-
compressed, err := kvGetter(dumpTable, base_encoding.Encode64ToBytes4(freshDumpSlot))
if err != nil {
return err
}
if len(compressed) != 0 {
- if _, err := buffer.Write(compressed); err != nil {
- return err
- }
- zstdReader, err := zstd.NewReader(buffer)
+ zstdReader, err := zstd.NewReader(bytes.NewReader(compressed))
if err != nil {
return err
}
@@ -1089,7 +1061,6 @@ func ReadQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors.GetValF
}
for currSlot := freshDumpSlot + 1; currSlot <= slot; currSlot++ {
- buffer.Reset()
key := base_encoding.Encode64ToBytes4(currSlot)
v, err := kvGetter(diffsTable, key)
if err != nil {
@@ -1099,11 +1070,7 @@ func ReadQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors.GetValF
continue
}
- if _, err := buffer.Write(v); err != nil {
- return err
- }
-
- if err := base_encoding.ApplySSZQueueDiff(buffer, out, 0); err != nil {
+ if err := base_encoding.ApplySSZQueueDiff(bytes.NewReader(v), out, 0); err != nil {
return err
}
}
@@ -1127,15 +1094,8 @@ func ReadRequiredQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors
return fmt.Errorf("%w: table %s, slot %d", ErrMissingGloasData, dumpTable, slot)
}
- // Decompress and decode the dump (reuse buffer pool).
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
- buffer.Reset()
-
- if _, err := buffer.Write(compressed); err != nil {
- return err
- }
- zstdReader, err := zstd.NewReader(buffer)
+ // Decompress and decode the dump.
+ zstdReader, err := zstd.NewReader(bytes.NewReader(compressed))
if err != nil {
return err
}
@@ -1150,7 +1110,6 @@ func ReadRequiredQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors
// Apply per-slot diffs from dump+1 to target slot.
for currSlot := freshDumpSlot + 1; currSlot <= slot; currSlot++ {
- buffer.Reset()
key := base_encoding.Encode64ToBytes4(currSlot)
v, err := kvGetter(diffsTable, key)
if err != nil {
@@ -1159,10 +1118,7 @@ func ReadRequiredQueueSSZ[T solid.EncodableHashableSSZ](kvGetter state_accessors
if len(v) == 0 {
continue
}
- if _, err := buffer.Write(v); err != nil {
- return err
- }
- if err := base_encoding.ApplySSZQueueDiff(buffer, out, 0); err != nil {
+ if err := base_encoding.ApplySSZQueueDiff(bytes.NewReader(v), out, 0); err != nil {
return err
}
}
@@ -1192,14 +1148,7 @@ func readCompressedSSZ[T interface {
return fmt.Errorf("%w: table %s, slot %d", ErrMissingGloasData, table, slot)
}
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
- buffer.Reset()
-
- if _, err := buffer.Write(compressed); err != nil {
- return err
- }
- zstdReader, err := zstd.NewReader(buffer)
+ zstdReader, err := zstd.NewReader(bytes.NewReader(compressed))
if err != nil {
return err
}
diff --git a/common/crypto/crypto.go b/common/crypto/crypto.go
index 9c363d94705..d23cdc40a0e 100644
--- a/common/crypto/crypto.go
+++ b/common/crypto/crypto.go
@@ -30,7 +30,6 @@ import (
"io"
"math/big"
"os"
- "sync"
keccak "github.com/erigontech/fastkeccak"
"github.com/holiman/uint256"
@@ -303,20 +302,10 @@ func PubkeyToAddress(p ecdsa.PublicKey) common.Address {
return common.BytesToAddress(Keccak256(pubBytes)[12:])
}
-// hasherPool holds LegacyKeccak hashers.
-var hasherPool = sync.Pool{
- New: func() any {
- return keccak.NewFastKeccak()
- },
-}
+// NewKeccakState returns a reset KeccakState from the pool shared with the common package.
+func NewKeccakState() keccak.KeccakState { return common.NewKeccakState() }
-// NewKeccakState creates a new KeccakState
-func NewKeccakState() keccak.KeccakState {
- h := hasherPool.Get().(keccak.KeccakState)
- h.Reset()
- return h
-}
-func ReturnToPool(h keccak.KeccakState) { hasherPool.Put(h) }
+func ReturnToPool(h keccak.KeccakState) { common.ReturnKeccakState(h) }
// FinalizeHash finalizes sha and returns a Keccak-256 digest as a value type,
// avoiding the heap escape that occurs when passing h[:] to an interface Read method.
diff --git a/common/hasher.go b/common/hasher.go
index 2940a6f437d..7f46e6ad943 100644
--- a/common/hasher.go
+++ b/common/hasher.go
@@ -26,30 +26,35 @@ type Hasher struct {
Sha keccak.KeccakState
}
-var hashersPool = sync.Pool{
+var keccakStatePool = sync.Pool{
New: func() any {
- return &Hasher{Sha: keccak.NewFastKeccak()}
+ return keccak.NewFastKeccak()
},
}
-func NewHasher() *Hasher {
- h := hashersPool.Get().(*Hasher)
- h.Sha.Reset()
- return h
+// NewKeccakState returns a reset KeccakState from a shared pool.
+func NewKeccakState() keccak.KeccakState {
+ sha := keccakStatePool.Get().(keccak.KeccakState)
+ sha.Reset()
+ return sha
}
-func ReturnHasherToPool(h *Hasher) { hashersPool.Put(h) }
+
+func ReturnKeccakState(sha keccak.KeccakState) { keccakStatePool.Put(sha) }
+
+func NewHasher() *Hasher { return &Hasher{Sha: NewKeccakState()} }
+func ReturnHasherToPool(h *Hasher) { ReturnKeccakState(h.Sha) }
func HashData(data []byte) (Hash, error) {
- h := NewHasher()
- defer ReturnHasherToPool(h)
+ sha := NewKeccakState()
+ defer ReturnKeccakState(sha)
- _, err := h.Sha.Write(data)
+ _, err := sha.Write(data)
if err != nil {
return Hash{}, err
}
var buf Hash
- _, err = h.Sha.Read(buf[:])
+ _, err = sha.Read(buf[:])
if err != nil {
return Hash{}, err
}
diff --git a/common/pool/pool.go b/common/pool/pool.go
new file mode 100644
index 00000000000..e114a921bf6
--- /dev/null
+++ b/common/pool/pool.go
@@ -0,0 +1,41 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package pool
+
+import (
+ "bytes"
+ "sync"
+)
+
+// Buffers that grew beyond maxBufferCap are dropped on Put so that one oversized use cannot pin memory in the pool.
+const maxBufferCap = 1 << 20
+
+var buffers = sync.Pool{New: func() any { return new(bytes.Buffer) }}
+
+// GetBuffer returns an empty buffer from the shared pool.
+func GetBuffer() *bytes.Buffer {
+ b := buffers.Get().(*bytes.Buffer)
+ b.Reset()
+ return b
+}
+
+func PutBuffer(b *bytes.Buffer) {
+ if b.Cap() > maxBufferCap {
+ return
+ }
+ buffers.Put(b)
+}
diff --git a/common/pool/pool_test.go b/common/pool/pool_test.go
new file mode 100644
index 00000000000..16e6086b29d
--- /dev/null
+++ b/common/pool/pool_test.go
@@ -0,0 +1,45 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package pool
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestGetBufferReturnsEmptyBuffer(t *testing.T) {
+ buf := GetBuffer()
+ buf.WriteString("dirty")
+ PutBuffer(buf)
+
+ fresh := GetBuffer()
+ defer PutBuffer(fresh)
+ require.Zero(t, fresh.Len())
+}
+
+func TestPutBufferDropsOversized(t *testing.T) {
+ buf := GetBuffer()
+ buf.Grow(maxBufferCap + 1)
+ require.Greater(t, buf.Cap(), maxBufferCap)
+ PutBuffer(buf)
+
+ // Any buffer obtained from the pool now must be within the cap limit.
+ fresh := GetBuffer()
+ defer PutBuffer(fresh)
+ require.LessOrEqual(t, fresh.Cap(), maxBufferCap)
+}
diff --git a/db/bufiopool/bufiopool.go b/db/bufiopool/bufiopool.go
new file mode 100644
index 00000000000..26adabed769
--- /dev/null
+++ b/db/bufiopool/bufiopool.go
@@ -0,0 +1,53 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+// Package bufiopool pools 512KB bufio readers and writers. Erigon doesn't
+// create tons of bufio readers/writers, but it has tons of parallel small
+// unit-tests which each create many small files and bufio readers/writers —
+// pooling avoids the allocation pressure in that scenario.
+package bufiopool
+
+import (
+ "bufio"
+ "io"
+ "sync"
+
+ "github.com/c2h5oh/datasize"
+)
+
+var (
+ writers = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(512*datasize.KB)) }}
+ readers = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, int(512*datasize.KB)) }}
+)
+
+func Writer(w io.Writer) *bufio.Writer {
+ bw := writers.Get().(*bufio.Writer)
+ bw.Reset(w)
+ return bw
+}
+
+// Reset(nil) before Put is required: without it the pool entry retains a
+// reference to the underlying io.Writer/io.Reader, keeping it alive until the
+// next GC cycle or until the entry is reused — whichever comes first.
+func PutWriter(w *bufio.Writer) { w.Reset(nil); writers.Put(w) }
+
+func Reader(r io.Reader) *bufio.Reader {
+ br := readers.Get().(*bufio.Reader)
+ br.Reset(r)
+ return br
+}
+
+func PutReader(r *bufio.Reader) { r.Reset(nil); readers.Put(r) }
diff --git a/db/datastruct/btindex/btree_index.go b/db/datastruct/btindex/btree_index.go
index 800af99f8d4..8f150d620bd 100644
--- a/db/datastruct/btindex/btree_index.go
+++ b/db/datastruct/btindex/btree_index.go
@@ -17,10 +17,8 @@
package btindex
import (
- "bufio"
"errors"
"fmt"
- "io"
"os"
"path"
"path/filepath"
@@ -28,7 +26,6 @@ import (
"time"
"unsafe"
- "github.com/c2h5oh/datasize"
"github.com/edsrzf/mmap-go"
"github.com/erigontech/erigon/common/background"
@@ -36,6 +33,7 @@ import (
"github.com/erigontech/erigon/common/dir"
"github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/common/murmur3"
+ "github.com/erigontech/erigon/db/bufiopool"
"github.com/erigontech/erigon/db/datastruct/existence"
"github.com/erigontech/erigon/db/recsplit/eliasfano32"
"github.com/erigontech/erigon/db/seg"
@@ -226,7 +224,7 @@ func NewBtIndexWriter(args BtIndexWriterArgs, logger log.Logger) (_ *BtIndexWrit
if btw.indexF, err = dir.CreateTemp(args.IndexFile); err != nil {
return nil, fmt.Errorf("create temp index file for %s: %w", args.IndexFile, err)
}
- btw.writer = &countingWriter{w: getBufioWriter(btw.indexF)}
+ btw.writer = &countingWriter{w: bufiopool.Writer(btw.indexF)}
if args.KeyCount > 0 {
if args.M == 0 {
@@ -352,7 +350,7 @@ func (btw *BtIndexWriter) closeTemps() {
func (btw *BtIndexWriter) Close() {
if btw.writer != nil {
- putBufioWriter(btw.writer.w)
+ bufiopool.PutWriter(btw.writer.w)
btw.writer = nil
}
if btw.indexF != nil { // non-nil means Build didn't rename it: drop the partial .tmp
@@ -724,19 +722,3 @@ func (b *BtIndex) OrdinalLookup(getter *seg.Reader, i uint64) *Cursor {
func (b *BtIndex) Offsets() *eliasfano32.EliasFano { return b.bplus.Offsets() }
func (b *BtIndex) Distances() (map[int]int, error) { return b.bplus.Distances() }
-
-// Erigon doesn't create tons of bufio readers/writers, but it has tons of
-// parallel small unit-tests which each create many small files and bufio
-// readers/writers — pooling avoids the allocation pressure in that scenario.
-var bufioWriterPool = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(512*datasize.KB)) }}
-
-func getBufioWriter(w io.Writer) *bufio.Writer {
- bw := bufioWriterPool.Get().(*bufio.Writer)
- bw.Reset(w)
- return bw
-}
-
-// Reset(nil) before Put is required: without it the pool entry retains a
-// reference to the underlying io.Writer/io.Reader, keeping it alive until the
-// next GC cycle or until the entry is reused — whichever comes first.
-func putBufioWriter(w *bufio.Writer) { w.Reset(nil); bufioWriterPool.Put(w) }
diff --git a/db/datastruct/btindex/btree_index_test.go b/db/datastruct/btindex/btree_index_test.go
index 60c893c61b1..f402a446289 100644
--- a/db/datastruct/btindex/btree_index_test.go
+++ b/db/datastruct/btindex/btree_index_test.go
@@ -31,6 +31,7 @@ import (
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/background"
"github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/db/bufiopool"
"github.com/erigontech/erigon/db/recsplit/eliasfano32"
"github.com/erigontech/erigon/db/seg"
"github.com/erigontech/erigon/db/state/statecfg"
@@ -205,8 +206,8 @@ func writeV0Index(tb testing.TB, dataPath, indexPath string, compressed seg.File
f, err := os.Create(indexPath)
require.NoError(tb, err)
defer f.Close()
- w := getBufioWriter(f)
- defer putBufioWriter(w)
+ w := bufiopool.Writer(f)
+ defer bufiopool.PutWriter(w)
if count > 0 {
ef := eliasfano32.NewEliasFano(count, uint64(r.Size()))
diff --git a/db/etl/dataprovider.go b/db/etl/dataprovider.go
index 30e7a73eac7..d6582864a22 100644
--- a/db/etl/dataprovider.go
+++ b/db/etl/dataprovider.go
@@ -17,21 +17,19 @@
package etl
import (
- "bufio"
"encoding/binary"
"fmt"
"io"
"os"
"path/filepath"
- "sync"
"sync/atomic"
- "github.com/c2h5oh/datasize"
"golang.org/x/sync/errgroup"
"github.com/erigontech/erigon/common/dir"
"github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/common/mmap"
+ "github.com/erigontech/erigon/db/bufiopool"
)
type dataProvider interface {
@@ -101,19 +99,6 @@ func FlushToDisk(logPrefix string, b Buffer, tmpdir string, lvl log.Lvl) (dataPr
return provider, nil
}
-var bufioWriterPool = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(512*datasize.KB)) }}
-
-func getBufioWriter(w io.Writer) *bufio.Writer {
- bw := bufioWriterPool.Get().(*bufio.Writer)
- bw.Reset(w)
- return bw
-}
-
-// Reset(nil) before Put is required: without it the pool entry retains a
-// reference to the underlying io.Writer/io.Reader, keeping it alive until the
-// next GC cycle or until the entry is reused — whichever comes first.
-func putBufioWriter(w *bufio.Writer) { w.Reset(nil); bufioWriterPool.Put(w) }
-
func sortAndFlush(b Buffer, tmpdir string) (*os.File, error) {
b.Sort()
@@ -130,8 +115,8 @@ func sortAndFlush(b Buffer, tmpdir string) (*os.File, error) {
return nil, err
}
- w := getBufioWriter(bufferFile)
- defer putBufioWriter(w)
+ w := bufiopool.Writer(bufferFile)
+ defer bufiopool.PutWriter(w)
if err = b.Write(w); err != nil {
return bufferFile, fmt.Errorf("error writing entries to disk: %w", err)
diff --git a/db/recsplit/recsplit.go b/db/recsplit/recsplit.go
index 0be13d255c4..f6aaa5bfb30 100644
--- a/db/recsplit/recsplit.go
+++ b/db/recsplit/recsplit.go
@@ -32,8 +32,6 @@ import (
"sync"
"time"
- "github.com/c2h5oh/datasize"
-
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/background"
"github.com/erigontech/erigon/common/dbg"
@@ -41,6 +39,7 @@ import (
"github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/common/mmap"
"github.com/erigontech/erigon/common/murmur3"
+ "github.com/erigontech/erigon/db/bufiopool"
"github.com/erigontech/erigon/db/datastruct/fusefilter"
"github.com/erigontech/erigon/db/etl"
"github.com/erigontech/erigon/db/recsplit/eliasfano16"
@@ -288,7 +287,7 @@ func NewRecSplit(args RecSplitArgs, logger log.Logger) (*RecSplit, error) {
if err != nil {
return nil, err
}
- rs.offsetWriter = getBufioWriter(rs.offsetFile)
+ rs.offsetWriter = bufiopool.Writer(rs.offsetFile)
}
if rs.enums && args.KeyCount > 0 && rs.lessFalsePositives {
if rs.dataStructureVersion == 0 {
@@ -296,7 +295,7 @@ func NewRecSplit(args RecSplitArgs, logger log.Logger) (*RecSplit, error) {
if err != nil {
return nil, err
}
- rs.existenceWV0 = getBufioWriter(rs.existenceFV0)
+ rs.existenceWV0 = bufiopool.Writer(rs.existenceFV0)
}
}
@@ -382,7 +381,7 @@ func (rs *RecSplit) Close() {
rs.existenceFV0 = nil
}
if rs.existenceWV0 != nil {
- putBufioWriter(rs.existenceWV0)
+ bufiopool.PutWriter(rs.existenceWV0)
rs.existenceWV0 = nil
}
if rs.existenceFV1 != nil {
@@ -402,7 +401,7 @@ func (rs *RecSplit) Close() {
rs.offsetFile = nil
}
if rs.offsetWriter != nil {
- putBufioWriter(rs.offsetWriter)
+ bufiopool.PutWriter(rs.offsetWriter)
rs.offsetWriter = nil
}
if rs.offsetEf != nil {
@@ -940,8 +939,8 @@ func (rs *RecSplit) Build(ctx context.Context) error {
}
defer rs.indexF.Close()
- rs.indexW = getBufioWriter(rs.indexF)
- defer putBufioWriter(rs.indexW)
+ rs.indexW = bufiopool.Writer(rs.indexF)
+ defer bufiopool.PutWriter(rs.indexW)
// 1 byte: dataStructureVersion, 7 bytes: app-specific minimal dataID (of current shard)
binary.BigEndian.PutUint64(rs.numBuf[:], rs.baseDataID)
rs.numBuf[0] = uint8(rs.dataStructureVersion)
@@ -1113,9 +1112,9 @@ func (rs *RecSplit) flushExistenceFilter() error {
if _, err := rs.existenceFV0.Seek(0, io.SeekStart); err != nil {
return err
}
- r := getBufioReader(rs.existenceFV0)
+ r := bufiopool.Reader(rs.existenceFV0)
_, copyErr := io.CopyN(rs.indexW, r, int64(rs.keysAdded))
- putBufioReader(r)
+ bufiopool.PutReader(r)
if copyErr != nil {
return copyErr
}
@@ -1472,29 +1471,3 @@ func (rs *RecSplit) buildWithWorkers(ctx context.Context) error {
}
return producerErr
}
-
-// Erigon doesn't create tons of bufio readers/writers, but it has tons of
-// parallel small unit-tests which each create many small files and bufio
-// readers/writers — pooling avoids the allocation pressure in that scenario.
-var (
- bufioWriterPool = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(512*datasize.KB)) }}
- bufioReaderPool = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, int(512*datasize.KB)) }}
-)
-
-func getBufioWriter(w io.Writer) *bufio.Writer {
- bw := bufioWriterPool.Get().(*bufio.Writer)
- bw.Reset(w)
- return bw
-}
-
-// Reset(nil) before Put is required: without it the pool entry retains a
-// reference to the underlying io.Writer/io.Reader, keeping it alive until the
-// next GC cycle or until the entry is reused — whichever comes first.
-func putBufioWriter(w *bufio.Writer) { w.Reset(nil); bufioWriterPool.Put(w) }
-
-func getBufioReader(r io.Reader) *bufio.Reader {
- br := bufioReaderPool.Get().(*bufio.Reader)
- br.Reset(r)
- return br
-}
-func putBufioReader(r *bufio.Reader) { r.Reset(nil); bufioReaderPool.Put(r) }
diff --git a/db/seg/compress.go b/db/seg/compress.go
index dd0a7e915f2..e4092c08208 100644
--- a/db/seg/compress.go
+++ b/db/seg/compress.go
@@ -33,12 +33,11 @@ import (
"sync"
"time"
- "github.com/c2h5oh/datasize"
-
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/dir"
dir2 "github.com/erigontech/erigon/common/dir"
"github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/db/bufiopool"
"github.com/erigontech/erigon/db/etl"
)
@@ -221,32 +220,6 @@ func (c *Compressor) SetMetadata(metadata []byte) {
func (c *Compressor) Count() int { return int(c.wordsCount) }
-// Erigon doesn't create tons of bufio readers/writers, but it has tons of
-// parallel small unit-tests which each create many small files and bufio
-// readers/writers — pooling avoids the allocation pressure in that scenario.
-var (
- bufioWriterPool = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(512*datasize.KB)) }}
- bufioReaderPool = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, int(512*datasize.KB)) }}
-)
-
-func getBufioWriter(w io.Writer) *bufio.Writer {
- bw := bufioWriterPool.Get().(*bufio.Writer)
- bw.Reset(w)
- return bw
-}
-
-// Reset(nil) before Put is required: without it the pool entry retains a
-// reference to the underlying io.Writer/io.Reader, keeping it alive until the
-// next GC cycle or until the entry is reused — whichever comes first.
-func putBufioWriter(w *bufio.Writer) { w.Reset(nil); bufioWriterPool.Put(w) }
-
-func getBufioReader(r io.Reader) *bufio.Reader {
- br := bufioReaderPool.Get().(*bufio.Reader)
- br.Reset(r)
- return br
-}
-func putBufioReader(r *bufio.Reader) { r.Reset(nil); bufioReaderPool.Put(r) }
-
func (c *Compressor) ReadFrom(g *Getter) error {
var v []byte
for g.HasNext() {
@@ -973,14 +946,14 @@ func NewRawWordsFile(filePath string) (*RawWordsFile, error) {
if err != nil {
return nil, err
}
- return &RawWordsFile{filePath: filePath, f: f, w: getBufioWriter(f), buf: make([]byte, 128)}, nil
+ return &RawWordsFile{filePath: filePath, f: f, w: bufiopool.Writer(f), buf: make([]byte, 128)}, nil
}
func OpenRawWordsFile(filePath string) (*RawWordsFile, error) {
f, err := os.Open(filePath)
if err != nil {
return nil, err
}
- return &RawWordsFile{filePath: filePath, f: f, w: getBufioWriter(f), buf: make([]byte, 128)}, nil
+ return &RawWordsFile{filePath: filePath, f: f, w: bufiopool.Writer(f), buf: make([]byte, 128)}, nil
}
func (f *RawWordsFile) Flush() error {
return f.w.Flush()
@@ -989,7 +962,7 @@ func (f *RawWordsFile) Close() {
if f.w != nil {
f.w.Flush()
f.f.Close()
- putBufioWriter(f.w)
+ bufiopool.PutWriter(f.w)
f.w = nil
f.f = nil
}
@@ -1033,8 +1006,8 @@ func (f *RawWordsFile) ForEach(walker func(v []byte, compressed bool) error) err
if err != nil {
return err
}
- r := getBufioReader(f.f)
- defer putBufioReader(r)
+ r := bufiopool.Reader(f.f)
+ defer bufiopool.PutReader(r)
buf := make([]byte, 16*1024)
l, e := binary.ReadUvarint(r)
for ; e == nil; l, e = binary.ReadUvarint(r) {
diff --git a/db/seg/parallel_compress.go b/db/seg/parallel_compress.go
index cdcf07fc010..bc0cf04d559 100644
--- a/db/seg/parallel_compress.go
+++ b/db/seg/parallel_compress.go
@@ -35,6 +35,7 @@ import (
"github.com/erigontech/erigon/common/dbg"
"github.com/erigontech/erigon/common/dir"
"github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/db/bufiopool"
"github.com/erigontech/erigon/db/etl"
"github.com/erigontech/erigon/db/seg/patricia"
"github.com/erigontech/erigon/db/seg/sais"
@@ -395,8 +396,8 @@ func compressWithPatternCandidates(ctx context.Context, trace bool, cfg Cfg, log
intermediatePath := intermediateFile.Name()
defer dir.RemoveFile(intermediatePath)
defer intermediateFile.Close()
- intermediateW := getBufioWriter(intermediateFile)
- defer putBufioWriter(intermediateW)
+ intermediateW := bufiopool.Writer(intermediateFile)
+ defer bufiopool.PutWriter(intermediateW)
var inCount, outCount, emptyWordsCount uint64 // Counters words sent to compression and returned for compression
var numBuf [binary.MaxVarintLen64]byte
@@ -634,8 +635,8 @@ func compressWithPatternCandidates(ctx context.Context, trace bool, cfg Cfg, log
if lvl < log.LvlTrace {
logger.Log(lvl, fmt.Sprintf("[%s] Effective dictionary", logPrefix), logCtx...)
}
- cw := getBufioWriter(cf)
- defer putBufioWriter(cw)
+ cw := bufiopool.Writer(cf)
+ defer bufiopool.PutWriter(cw)
// 1-st, output amount of words - just a useful metadata
binary.BigEndian.PutUint64(numBuf[:], inCount) // Dictionary size
if _, err = cw.Write(numBuf[:8]); err != nil {
@@ -693,8 +694,8 @@ func compressWithPatternCandidates(ctx context.Context, trace bool, cfg Cfg, log
wc := 0
var hc BitWriter
hc.w = cw
- r := getBufioReader(intermediateFile)
- defer putBufioReader(r)
+ r := bufiopool.Reader(intermediateFile)
+ defer bufiopool.PutReader(r)
copyNBuf := make([]byte, 32*1024)
var l uint64
@@ -810,8 +811,8 @@ func compressNoWordPatterns(logPrefix string, cf *os.File, uncompressedFile *Raw
return err
}
- cw := getBufioWriter(cf)
- defer putBufioWriter(cw)
+ cw := bufiopool.Writer(cf)
+ defer bufiopool.PutWriter(cw)
// Write data header: word count, empty word count, patternsSize=0, then position dict.
binary.BigEndian.PutUint64(numBuf[:], inCount)
@@ -1174,8 +1175,8 @@ func PersistDictionary(fileName string, db *DictionaryBuilder) error {
if err != nil {
return err
}
- w := getBufioWriter(df)
- defer putBufioWriter(w)
+ w := bufiopool.Writer(df)
+ defer bufiopool.PutWriter(w)
db.ForEach(func(score uint64, word []byte) { fmt.Fprintf(w, "%d %x\n", score, word) })
if err = w.Flush(); err != nil {
return err
@@ -1194,8 +1195,8 @@ func ReadSimpleFile(fileName string, walker func(v []byte) error) error {
return err
}
defer f.Close()
- r := getBufioReader(f)
- defer putBufioReader(r)
+ r := bufiopool.Reader(f)
+ defer bufiopool.PutReader(r)
buf := make([]byte, 4096)
for l, e := binary.ReadUvarint(r); ; l, e = binary.ReadUvarint(r) {
if e != nil {
diff --git a/db/snapshotsync/freezeblocks/beacon_block_reader.go b/db/snapshotsync/freezeblocks/beacon_block_reader.go
index 9a25d5ecb3c..35b562cb941 100644
--- a/db/snapshotsync/freezeblocks/beacon_block_reader.go
+++ b/db/snapshotsync/freezeblocks/beacon_block_reader.go
@@ -34,10 +34,6 @@ import (
"github.com/erigontech/erigon/execution/types"
)
-var buffersPool = sync.Pool{
- New: func() any { return &bytes.Buffer{} },
-}
-
var decompressorPool = sync.Pool{
New: func() any {
r, err := zstd.NewReader(nil)
@@ -135,16 +131,11 @@ func (r *beaconSnapshotReader) ReadBlockBySlot(ctx context.Context, tx kv.Tx, sl
}
// Decompress this thing
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
-
- buffer.Reset()
- buffer.Write(buf)
reader := decompressorPool.Get().(*zstd.Decoder)
defer decompressorPool.Put(reader)
- reader.Reset(buffer)
+ reader.Reset(bytes.NewReader(buf))
- // Use pooled buffers and readers to avoid allocations.
+ // Use pooled readers to avoid allocations.
return snapshot_format.ReadBlockFromSnapshot(reader, r.eth1Getter, r.cfg)
}
@@ -197,16 +188,11 @@ func (r *beaconSnapshotReader) ReadBeaconBlockBodyBySlot(ctx context.Context, tx
}
// Decompress this thing
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
-
- buffer.Reset()
- buffer.Write(buf)
reader := decompressorPool.Get().(*zstd.Decoder)
defer decompressorPool.Put(reader)
- reader.Reset(buffer)
+ reader.Reset(bytes.NewReader(buf))
- // Use pooled buffers and readers to avoid allocations.
+ // Use pooled readers to avoid allocations.
return snapshot_format.ReadBeaconBlockBodyFromSnapshot(reader, r.cfg)
}
@@ -275,16 +261,11 @@ func (r *beaconSnapshotReader) ReadBlockByRoot(ctx context.Context, tx kv.Tx, ro
return nil, nil
}
// Decompress this thing
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
-
- buffer.Reset()
- buffer.Write(buf)
reader := decompressorPool.Get().(*zstd.Decoder)
defer decompressorPool.Put(reader)
- reader.Reset(buffer)
+ reader.Reset(bytes.NewReader(buf))
- // Use pooled buffers and readers to avoid allocations.
+ // Use pooled readers to avoid allocations.
return snapshot_format.ReadBlockFromSnapshot(reader, r.eth1Getter, r.cfg)
}
@@ -315,6 +296,5 @@ func (r *beaconSnapshotReader) ReadHeaderByRoot(ctx context.Context, tx kv.Tx, r
}
h, _, _, err := r.sn.ReadHeader(*slot, tx)
- // Use pooled buffers and readers to avoid allocations.
return h, err
}
diff --git a/db/snapshotsync/freezeblocks/caplin_snapshots.go b/db/snapshotsync/freezeblocks/caplin_snapshots.go
index 3ffd266540f..e6866ee1d0e 100644
--- a/db/snapshotsync/freezeblocks/caplin_snapshots.go
+++ b/db/snapshotsync/freezeblocks/caplin_snapshots.go
@@ -713,16 +713,11 @@ func (s *CaplinSnapshots) ReadHeader(slot uint64, tx kv.Tx) (*cltypes.SignedBeac
return nil, 0, common.Hash{}, nil
}
// Decompress this thing
- buffer := buffersPool.Get().(*bytes.Buffer)
- defer buffersPool.Put(buffer)
-
- buffer.Reset()
- buffer.Write(buf)
reader := decompressorPool.Get().(*zstd.Decoder)
defer decompressorPool.Put(reader)
- reader.Reset(buffer)
+ reader.Reset(bytes.NewReader(buf))
- // Use pooled buffers and readers to avoid allocations.
+ // Use pooled readers to avoid allocations.
header, elBlockNumber, elBlockHash, err := snapshot_format.ReadBlockHeaderFromSnapshotWithExecutionData(reader, s.beaconCfg)
if err != nil {
return nil, 0, common.Hash{}, err
diff --git a/execution/types/hashing.go b/execution/types/hashing.go
index 7653fdfa7d1..83eaedca38a 100644
--- a/execution/types/hashing.go
+++ b/execution/types/hashing.go
@@ -23,7 +23,6 @@ import (
"bytes"
"fmt"
"io"
- "sync"
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/crypto"
@@ -31,11 +30,6 @@ import (
"github.com/erigontech/erigon/execution/rlp"
)
-// encodeBufferPool holds temporary encoder buffers for DeriveSha and TX encoding.
-var encodeBufferPool = sync.Pool{
- New: func() any { return new(bytes.Buffer) },
-}
-
type DerivableList interface {
Len() int
EncodeIndex(i int, w *bytes.Buffer)
diff --git a/execution/types/receipt.go b/execution/types/receipt.go
index 2e2630378cb..87e0b0c476e 100644
--- a/execution/types/receipt.go
+++ b/execution/types/receipt.go
@@ -32,6 +32,7 @@ import (
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/dbg"
"github.com/erigontech/erigon/common/hexutil"
+ "github.com/erigontech/erigon/common/pool"
"github.com/erigontech/erigon/execution/rlp"
)
@@ -145,9 +146,8 @@ func (r Receipt) EncodeRLP(w io.Writer) error {
if r.Type == LegacyTxType {
return rlp.Encode(w, data)
}
- buf := encodeBufferPool.Get().(*bytes.Buffer)
- defer encodeBufferPool.Put(buf)
- buf.Reset()
+ buf := pool.GetBuffer()
+ defer pool.PutBuffer(buf)
if err := r.encodeTyped(data, buf); err != nil {
return err
}
diff --git a/node/rpcstack.go b/node/rpcstack.go
index 8aaa6858bbf..a90258f1f39 100644
--- a/node/rpcstack.go
+++ b/node/rpcstack.go
@@ -36,6 +36,7 @@ import (
"github.com/rs/cors"
"github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/common/pool"
"github.com/erigontech/erigon/db/kv"
"github.com/erigontech/erigon/diagnostics/metrics"
"github.com/erigontech/erigon/rpc"
@@ -182,7 +183,7 @@ func (h *virtualHostHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "invalid host specified", http.StatusForbidden)
}
-// gzPoolBufCap is the maximum buffer capacity retained in pools to bound RSS growth.
+// gzPoolBufCap is the maximum dst capacity retained in the pool to bound RSS growth.
const gzPoolBufCap = 1 << 20
// minGzipBodySize is the minimum response body size to compress. Responses
@@ -211,26 +212,10 @@ var gzCompressorPool = sync.Pool{
},
}
-var gzBufPool = sync.Pool{
- New: func() any { return new(bytes.Buffer) },
-}
-
var gzDstPool = sync.Pool{
New: func() any { return make([]byte, 0, 64*1024) },
}
-func getBuf() *bytes.Buffer {
- buf := gzBufPool.Get().(*bytes.Buffer)
- buf.Reset()
- return buf
-}
-
-func putBuf(buf *bytes.Buffer) {
- if buf.Cap() <= gzPoolBufCap {
- gzBufPool.Put(buf)
- }
-}
-
func putDst(dst []byte) {
if cap(dst) <= gzPoolBufCap {
gzDstPool.Put(dst)
@@ -330,7 +315,7 @@ func compressLibdeflate(w http.ResponseWriter, src []byte, status int) bool {
}
func sendGzipResponse(w http.ResponseWriter, grw *gzipResponseWriter) {
- defer putBuf(grw.buf)
+ defer pool.PutBuffer(grw.buf)
if grw.gzw != nil {
defer gzPool.Put(grw.gzw)
@@ -360,7 +345,7 @@ func newGzipHandler(next http.Handler) http.Handler {
return
}
- grw := &gzipResponseWriter{buf: getBuf(), ResponseWriter: w}
+ grw := &gzipResponseWriter{buf: pool.GetBuffer(), ResponseWriter: w}
// The hook activates streaming mode before the first write; absent when gzip
// is off so it cannot prematurely commit HTTP headers (e.g. 200 before 503).
r = r.WithContext(rpc.WithGzipStreamingHook(r.Context(), grw.Flush))
diff --git a/node/rpcstack_gzip_handler_test.go b/node/rpcstack_gzip_handler_test.go
index 10365a3dae7..fdb2678e787 100644
--- a/node/rpcstack_gzip_handler_test.go
+++ b/node/rpcstack_gzip_handler_test.go
@@ -155,7 +155,7 @@ func TestGzipHandlerStatusStreaming(t *testing.T) {
// TestGzipHandlerLargeBody verifies that a response body larger than
// gzPoolBufCap (1 MiB) is compressed correctly. This exercises the pool-cap
-// path: the oversized buffer must not be returned to gzBufPool.
+// path: the oversized buffer must not be returned to the shared buffer pool.
func TestGzipHandlerLargeBody(t *testing.T) {
body := bytes.Repeat([]byte("x"), gzPoolBufCap+1)
handler := newGzipHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -173,8 +173,7 @@ func TestGzipHandlerLargeBody(t *testing.T) {
// mode and that buffered bytes are drained into the gzip writer.
func TestGzipResponseWriterFlushActivatesStreaming(t *testing.T) {
rec := httptest.NewRecorder()
- buf := gzBufPool.Get().(*bytes.Buffer)
- buf.Reset()
+ buf := &bytes.Buffer{}
grw := &gzipResponseWriter{buf: buf, ResponseWriter: rec}
// Write into the buffer before activating streaming.
@@ -193,22 +192,3 @@ func TestGzipResponseWriterFlushActivatesStreaming(t *testing.T) {
got := decompressGzip(t, rec.Body)
assert.Equal(t, []byte("pre-flush post-flush"), got)
}
-
-// TestGzipBufPoolCapThreshold verifies that the cap guard does not return
-// an oversized buffer to gzBufPool.
-func TestGzipBufPoolCapThreshold(t *testing.T) {
- buf := gzBufPool.Get().(*bytes.Buffer)
- buf.Grow(gzPoolBufCap + 1)
- require.Greater(t, buf.Cap(), gzPoolBufCap)
-
- // Simulate the handler's cap check: large buffer is NOT returned.
- if buf.Cap() <= gzPoolBufCap {
- gzBufPool.Put(buf)
- }
-
- // Any buffer obtained from the pool now must be within the cap limit.
- fresh := gzBufPool.Get().(*bytes.Buffer)
- defer gzBufPool.Put(fresh)
- assert.LessOrEqual(t, fresh.Cap(), gzPoolBufCap,
- "pool must not contain a buffer larger than gzPoolBufCap")
-}
diff --git a/rpc/requests/request_generator.go b/rpc/requests/request_generator.go
index a2841b0e171..8eb5f8391c0 100644
--- a/rpc/requests/request_generator.go
+++ b/rpc/requests/request_generator.go
@@ -29,6 +29,7 @@ import (
"sync"
"time"
+ "github.com/cenkalti/backoff/v4"
"github.com/valyala/fastjson"
"github.com/erigontech/erigon/common"
@@ -234,41 +235,38 @@ func isConnectionError(err error) bool {
func retryConnects(ctx context.Context, op func(context.Context) error) error {
ctx, cancel := context.WithTimeout(ctx, requestTimeout)
defer cancel()
- return retry(ctx, op, isConnectionError, time.Second*1, nil)
-}
-
-func retry(ctx context.Context, op func(context.Context) error, isRecoverableError func(error) bool, delay time.Duration, lastErr error) error {
- opctx, cancel := context.WithTimeout(ctx, connectionTimeout)
- defer cancel()
-
- err := op(opctx)
- if err == nil {
- return nil
- }
-
- if !isRecoverableError(err) {
- return err
- }
+ var lastDialErr error
+ attempt := func() error {
+ opctx, opCancel := context.WithTimeout(ctx, connectionTimeout)
+ defer opCancel()
- if errors.Is(err, context.DeadlineExceeded) {
- if lastErr != nil {
- return lastErr
+ err := op(opctx)
+ if err == nil {
+ return nil
}
-
- err = nil
- }
-
- delayTimer := time.NewTimer(delay)
- select {
- case <-delayTimer.C:
- return retry(ctx, op, isRecoverableError, delay, err)
- case <-ctx.Done():
- if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+ if !isConnectionError(err) {
+ return backoff.Permanent(err)
+ }
+ if errors.Is(err, context.DeadlineExceeded) {
+ if lastDialErr != nil {
+ return backoff.Permanent(lastDialErr)
+ }
return err
}
- return ctx.Err()
+ lastDialErr = err
+ return err
+ }
+
+ err := backoff.Retry(attempt, backoff.WithContext(backoff.NewConstantBackOff(time.Second), ctx))
+ // backoff.Retry surfaces the overall context's error verbatim when it gives
+ // up; on deadline expiry, report the last dial error it discarded. Compared
+ // by identity so a permanent error that merely wraps DeadlineExceeded (from
+ // backoff.Permanent) is left intact.
+ if lastDialErr != nil && err == context.DeadlineExceeded {
+ return lastDialErr
}
+ return err
}
type PingResult callResult
diff --git a/rpc/requests/retry_test.go b/rpc/requests/retry_test.go
new file mode 100644
index 00000000000..4ab4dfdd7ff
--- /dev/null
+++ b/rpc/requests/retry_test.go
@@ -0,0 +1,154 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package requests
+
+import (
+ "context"
+ "errors"
+ "net"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func newDialError() *net.OpError {
+ return &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}
+}
+
+func sequenceOp(attempts *atomic.Int64, errs ...error) func(context.Context) error {
+ return func(context.Context) error {
+ n := int(attempts.Add(1))
+ if n > len(errs) {
+ return errs[len(errs)-1]
+ }
+ return errs[n-1]
+ }
+}
+
+func TestRetryConnectsSuccess(t *testing.T) {
+ t.Parallel()
+ var attempts atomic.Int64
+ err := retryConnects(t.Context(), sequenceOp(&attempts, nil))
+ require.NoError(t, err)
+ require.EqualValues(t, 1, attempts.Load())
+}
+
+func TestRetryConnectsNonRecoverableErrorNotRetried(t *testing.T) {
+ t.Parallel()
+ boom := errors.New("boom")
+ var attempts atomic.Int64
+ err := retryConnects(t.Context(), sequenceOp(&attempts, boom))
+ require.ErrorIs(t, err, boom)
+ require.EqualValues(t, 1, attempts.Load())
+}
+
+func TestRetryConnectsDialErrorRetriedUntilSuccess(t *testing.T) {
+ t.Parallel()
+ var attempts atomic.Int64
+ start := time.Now()
+ err := retryConnects(t.Context(), sequenceOp(&attempts, newDialError(), nil))
+ require.NoError(t, err)
+ require.EqualValues(t, 2, attempts.Load())
+ require.GreaterOrEqual(t, time.Since(start), 900*time.Millisecond)
+}
+
+func TestRetryConnectsTimeoutAfterDialErrorReturnsDialError(t *testing.T) {
+ t.Parallel()
+ dialErr := newDialError()
+ var attempts atomic.Int64
+ err := retryConnects(t.Context(), sequenceOp(&attempts, dialErr, context.DeadlineExceeded))
+ require.ErrorIs(t, err, dialErr)
+ require.EqualValues(t, 2, attempts.Load())
+}
+
+func TestRetryConnectsTimeoutRetriedUntilSuccess(t *testing.T) {
+ t.Parallel()
+ var attempts atomic.Int64
+ err := retryConnects(t.Context(), sequenceOp(&attempts, context.DeadlineExceeded, nil))
+ require.NoError(t, err)
+ require.EqualValues(t, 2, attempts.Load())
+}
+
+func TestRetryConnectsParentDeadlineReturnsLastDialError(t *testing.T) {
+ t.Parallel()
+ ctx, cancel := context.WithTimeout(t.Context(), 1200*time.Millisecond)
+ defer cancel()
+ dialErr := newDialError()
+ var attempts atomic.Int64
+ err := retryConnects(ctx, sequenceOp(&attempts, dialErr))
+ require.ErrorIs(t, err, dialErr)
+}
+
+func TestRetryConnectsParentDeadlineAfterOnlyTimeoutsReturnsDeadlineExceeded(t *testing.T) {
+ t.Parallel()
+ ctx, cancel := context.WithTimeout(t.Context(), 1200*time.Millisecond)
+ defer cancel()
+ var attempts atomic.Int64
+ err := retryConnects(ctx, sequenceOp(&attempts, context.DeadlineExceeded))
+ require.ErrorIs(t, err, context.DeadlineExceeded)
+}
+
+func TestRetryConnectsPermanentErrorWrappingDeadlineExceededNotSwallowed(t *testing.T) {
+ t.Parallel()
+ readErr := &net.OpError{Op: "read", Net: "tcp", Err: errors.New("connection reset")}
+ var attempts atomic.Int64
+ err := retryConnects(t.Context(), sequenceOp(&attempts, errors.Join(readErr, context.DeadlineExceeded)))
+ require.ErrorIs(t, err, readErr)
+ require.EqualValues(t, 1, attempts.Load())
+}
+
+func TestRetryConnectsParentCancel(t *testing.T) {
+ t.Parallel()
+ ctx, cancel := context.WithCancel(t.Context())
+ go func() {
+ time.Sleep(100 * time.Millisecond)
+ cancel()
+ }()
+ var attempts atomic.Int64
+ err := retryConnects(ctx, sequenceOp(&attempts, newDialError()))
+ require.ErrorIs(t, err, context.Canceled)
+}
+
+// TestRetryConnectsOverallDeadlineDoesNotClobberPermanentError pins that on
+// overall-deadline expiry the last dial error replaces only the context's own
+// deadline error, never a permanent error that merely wraps DeadlineExceeded.
+func TestRetryConnectsOverallDeadlineDoesNotClobberPermanentError(t *testing.T) {
+ t.Parallel()
+ // Deadline lands after the first 1s backoff (so a second attempt runs) but
+ // within the per-attempt window (so the second op observes it via its ctx).
+ ctx, cancel := context.WithTimeout(t.Context(), 1400*time.Millisecond)
+ defer cancel()
+
+ dialErr := newDialError()
+ readErr := &net.OpError{Op: "read", Net: "tcp", Err: errors.New("connection reset")}
+ permanentErr := errors.Join(readErr, context.DeadlineExceeded)
+ var attempts atomic.Int64
+ op := func(opctx context.Context) error {
+ if attempts.Add(1) == 1 {
+ return dialErr // recoverable → recorded as lastDialErr
+ }
+ <-opctx.Done() // let the overall deadline expire before returning
+ return permanentErr
+ }
+
+ err := retryConnects(ctx, op)
+ require.EqualValues(t, 2, attempts.Load())
+ require.ErrorIs(t, err, readErr)
+ require.NotErrorIs(t, err, dialErr)
+}