cherry-pick 14 commits from release/3.4 to main

.claude/skills/erigon-exec-from-0/SKILL.md

@@ -1,27 +1,70 @@
 ---
 name: erigon-exec-from-0
-description: Re exec all blocks from 0 to reproduce bugs
+description: >
+  Re-execute all Ethereum blocks from block 0 to reproduce bugs, validate state, or test
+  correctness after code changes. Use this skill whenever the user wants to re-run execution
+  from scratch, reset execution stage, reproduce a state mismatch, validate a fix by re-exec,
+  or run blocks from the beginning on a specific chain/datadir.
 ---
 
+# Re-execute all blocks from 0
+
+Use this workflow to wipe execution state and re-run all blocks from block 0.
+Common reasons: reproducing a state root mismatch, validating a bug fix, testing
+new execution logic end-to-end.
+
+## Prerequisites
+
+You need:
+- `--datadir` — path to the Erigon data directory (block snapshots must be present)
+- `--chain` — chain name (e.g. `chiado`, `mainnet`, `gnosis`, `sepolia`)
+
+## Step 1: Remove state snapshots (keep block snapshots)
+
+```bash
+go run ./cmd/erigon snapshots rm-all-state-snapshots --datadir <datadir>
 ```
-# remove all State files (not Block files)
-go run ./cmd/erigon snapshots rm-all-state-snapshots --datadir ~/data/chiado34_regen
 
-# clean database tables related to stage_exec
-go run ./cmd/integration stage_exec --datadir ~/data/chiado34_regen --chain=chiado --reset
+This removes `.kv` state snapshot files but leaves `.seg` block snapshot files intact,
+so you don't need to re-download block data.
 
-# exec blocks
-go run ./cmd/integration stage_exec --datadir ~/data/chiado34_regen --chain=chiado --batchSize=8m
+## Step 2: Reset the execution stage
 
-# flags for profiler
+```bash
+go run ./cmd/integration stage_exec --datadir <datadir> --chain=<chain> --reset
+```
+
+This clears the execution-stage progress in the database so the next run starts from block 0.
+
+## Step 3: Run execution
+
+```bash
+go run ./cmd/integration stage_exec --datadir <datadir> --chain=<chain> --batchSize=8m
+```
+
+Adjust `--batchSize` to control how frequently state is flushed to disk. Larger = faster
+but more RAM; `8m` is a reasonable default.
+
+## Useful optional flags
+
+**Profiling** (add when you want pprof):
+```
 --pprof --pprof.port=6160
+```
 
-# flags for correctness
-`ERIGON_ASSERT=true`
+**Correctness assertions** (add when hunting state bugs — slows execution):
+```
+ERIGON_ASSERT=true
+```
+
+**Stop at a specific block** (useful to reproduce a bug at block N):
+```
+--block=<N>
 ```
 
-### tools for files correctness check
+## Verifying snapshot integrity
 
+If you suspect corrupted snapshot files before running:
+```bash
+go run ./cmd/erigon seg integrity --datadir <datadir> --sample=0.001
 ```
-go run ./cmd/erigon seg integrity --datadir ~/data/chiado34_archive --sample=0.001 
-```

cl/antiquary/beacon_states_collector.go

@@ -356,75 +356,77 @@ func (i *beaconStatesCollector) collectInactivityScores(slot uint64, inactivityS
 }
 
 func (i *beaconStatesCollector) flush(ctx context.Context, tx kv.RwTx) error {
-	if err := i.effectiveBalanceCollector.Load(tx, kv.ValidatorEffectiveBalance, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	loadfunc := etl.IdentityLoadFunc
+
+	if err := i.effectiveBalanceCollector.Load(tx, kv.ValidatorEffectiveBalance, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.randaoMixesCollector.Load(tx, kv.RandaoMixes, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.randaoMixesCollector.Load(tx, kv.RandaoMixes, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.balancesCollector.Load(tx, kv.ValidatorBalance, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.balancesCollector.Load(tx, kv.ValidatorBalance, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
 
-	if err := i.slashingsCollector.Load(tx, kv.ValidatorSlashings, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.slashingsCollector.Load(tx, kv.ValidatorSlashings, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.blockRootsCollector.Load(tx, kv.BlockRoot, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.blockRootsCollector.Load(tx, kv.BlockRoot, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.stateRootsCollector.Load(tx, kv.StateRoot, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.stateRootsCollector.Load(tx, kv.StateRoot, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.activeValidatorIndiciesCollector.Load(tx, kv.ActiveValidatorIndicies, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.activeValidatorIndiciesCollector.Load(tx, kv.ActiveValidatorIndicies, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.slotDataCollector.Load(tx, kv.SlotData, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.slotDataCollector.Load(tx, kv.SlotData, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.inactivityScoresCollector.Load(tx, kv.InactivityScores, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.inactivityScoresCollector.Load(tx, kv.InactivityScores, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.intraRandaoMixesCollector.Load(tx, kv.IntraRandaoMixes, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.intraRandaoMixesCollector.Load(tx, kv.IntraRandaoMixes, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.epochDataCollector.Load(tx, kv.EpochData, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.epochDataCollector.Load(tx, kv.EpochData, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.nextSyncCommitteeCollector.Load(tx, kv.NextSyncCommittee, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.nextSyncCommitteeCollector.Load(tx, kv.NextSyncCommittee, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.currentSyncCommitteeCollector.Load(tx, kv.CurrentSyncCommittee, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.currentSyncCommitteeCollector.Load(tx, kv.CurrentSyncCommittee, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.eth1DataVotesCollector.Load(tx, kv.Eth1DataVotes, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.eth1DataVotesCollector.Load(tx, kv.Eth1DataVotes, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.stateEventsCollector.Load(tx, kv.StateEvents, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.stateEventsCollector.Load(tx, kv.StateEvents, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.effectiveBalancesDumpCollector.Load(tx, kv.EffectiveBalancesDump, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.effectiveBalancesDumpCollector.Load(tx, kv.EffectiveBalancesDump, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingDepositsCollector.Load(tx, kv.PendingDeposits, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingDepositsCollector.Load(tx, kv.PendingDeposits, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingConsolidationsCollector.Load(tx, kv.PendingConsolidations, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingConsolidationsCollector.Load(tx, kv.PendingConsolidations, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingWithdrawalsCollector.Load(tx, kv.PendingPartialWithdrawals, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingWithdrawalsCollector.Load(tx, kv.PendingPartialWithdrawals, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingDepositsCollectorDump.Load(tx, kv.PendingDepositsDump, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingDepositsCollectorDump.Load(tx, kv.PendingDepositsDump, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingConsolidationsCollectorDump.Load(tx, kv.PendingConsolidationsDump, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingConsolidationsCollectorDump.Load(tx, kv.PendingConsolidationsDump, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
-	if err := i.pendingWithdrawalsCollectorDump.Load(tx, kv.PendingPartialWithdrawalsDump, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
+	if err := i.pendingWithdrawalsCollectorDump.Load(tx, kv.PendingPartialWithdrawalsDump, loadfunc, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
 
-	return i.balancesDumpsCollector.Load(tx, kv.BalancesDump, etl.IdentityLoadFunc, etl.TransformArgs{Quit: ctx.Done()})
+	return i.balancesDumpsCollector.Load(tx, kv.BalancesDump, loadfunc, etl.TransformArgs{Quit: ctx.Done()})
 }
 
 func (i *beaconStatesCollector) close() {

cl/phase1/execution_client/block_collector/persistent_block_collector.go

@@ -205,8 +205,6 @@ func (p *PersistentBlockCollector) Flush(ctx context.Context) error {
 	}
 	p.db = db
 
-	p.logger.Info("[BlockCollector] Flush complete", "blocksInserted", inserted)
-
 	return nil
 }
 

db/etl/collector.go

@@ -173,7 +173,6 @@ func (c *Collector) Load(db kv.RwTx, toBucket string, loadFunc LoadFunc, args Tr
 	if c.buf == nil && c.allocator != nil {
 		c.buf = c.allocator.Get()
 	}
-	defer c.Close()
 	args.BufferType = c.bufType
 
 	if !c.allFlushed {

db/etl/dataprovider.go

@@ -23,8 +23,10 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sync"
 	"sync/atomic"
 
+	"github.com/c2h5oh/datasize"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/erigontech/erigon/common/dir"
@@ -99,6 +101,19 @@ 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()
 
@@ -115,12 +130,15 @@ func sortAndFlush(b Buffer, tmpdir string) (*os.File, error) {
 		return nil, err
 	}
 
-	w := bufio.NewWriterSize(bufferFile, BufIOSize)
-	defer w.Flush() //nolint:errcheck
+	w := getBufioWriter(bufferFile)
+	defer putBufioWriter(w)
 
 	if err = b.Write(w); err != nil {
 		return bufferFile, fmt.Errorf("error writing entries to disk: %w", err)
 	}
+	if err = w.Flush(); err != nil {
+		return bufferFile, fmt.Errorf("error flushing buffer to disk: %w", err)
+	}
 	return bufferFile, nil
 }
 
@@ -201,15 +219,17 @@ func (p *fileDataProvider) Dispose() {
 
 	p.Wait()
 
+	if p.mmapData != nil {
+		_ = mmap.Munmap(p.mmapData, p.mmapHandle2)
+		p.mmapData = nil
+		p.mmapHandle2 = nil
+		p.mmapReader = nil
+	}
+
 	filePath := p.file.Name()
 	p.file.Close()
 	p.file = nil
 	_ = dir.RemoveFile(filePath)
-
-	// Note: We intentionally do NOT munmap here. The mmap'd memory remains mapped
-	// and valid for zero-copy slices returned to callers. The OS will unmap when
-	// the process exits or memory pressure requires it. This is safe for the ETL
-	// use case where data is consumed immediately before Close() is called.
 }
 
 func (p *fileDataProvider) String() string {

db/etl/etl_test.go

@@ -485,6 +485,7 @@ func TestReuseCollectorAfterLoad(t *testing.T) {
 	}, TransformArgs{})
 	require.NoError(t, err)
 	require.Equal(t, 1, see)
+	c.Close()
 
 	// buffers are not lost
 	require.Empty(t, buf.data)
@@ -500,6 +501,7 @@ func TestReuseCollectorAfterLoad(t *testing.T) {
 	}, TransformArgs{})
 	require.NoError(t, err)
 	require.Equal(t, 0, see)
+	c.Close()
 
 	// reuse
 	see = 0
@@ -509,6 +511,7 @@ func TestReuseCollectorAfterLoad(t *testing.T) {
 	}, TransformArgs{})
 	require.NoError(t, err)
 	require.Equal(t, 0, see)
+	c.Close()
 
 	err = c.Collect([]byte{3}, []byte{4})
 	require.NoError(t, err)

db/kv/bitmapdb/bitmapdb.go

@@ -19,6 +19,7 @@ package bitmapdb
 import (
 	"sync"
 
+	"github.com/RoaringBitmap/roaring/v2"
 	"github.com/RoaringBitmap/roaring/v2/roaring64"
 )
 
@@ -39,3 +40,24 @@ func ReturnToPool64(a *roaring64.Bitmap) {
 	}
 	roaring64Pool.Put(a)
 }
+
+var roaring32Pool = sync.Pool{
+	New: func() any {
+		return roaring.New()
+	},
+}
+
+// NewBitmap32 returns a pooled 32-bit roaring bitmap.
+// Use when values fit in uint32 (e.g. txNum offsets within a collate step).
+func NewBitmap32() *roaring.Bitmap {
+	a := roaring32Pool.Get().(*roaring.Bitmap)
+	a.Clear()
+	return a
+}
+
+func ReturnToPool32(a *roaring.Bitmap) {
+	if a == nil {
+		return
+	}
+	roaring32Pool.Put(a)
+}

db/recsplit/recsplit.go

@@ -260,7 +260,7 @@ func NewRecSplit(args RecSplitArgs, logger log.Logger) (*RecSplit, error) {
 	} else {
 		rs.salt = *args.Salt
 	}
-	rs.bucketCollector = etl.NewCollectorWithAllocator(RecSplitLogPrefix+" "+fname, rs.tmpDir, etl.SmallSortableBuffers, logger)
+	rs.bucketCollector = etl.NewCollectorWithAllocator(RecSplitLogPrefix+" "+fname, rs.tmpDir, etl.LargeSortableBuffers, logger)
 	rs.bucketCollector.SortAndFlushInBackground(rs.workers > 1)
 	rs.bucketCollector.LogLvl(log.LvlDebug)
 	var err error

db/seg/compress.go

@@ -34,10 +34,8 @@ import (
 	"time"
 
 	"github.com/c2h5oh/datasize"
-	"golang.org/x/sync/semaphore"
 
 	"github.com/erigontech/erigon/common"
-	"github.com/erigontech/erigon/common/dbg"
 	"github.com/erigontech/erigon/common/dir"
 	dir2 "github.com/erigontech/erigon/common/dir"
 	"github.com/erigontech/erigon/common/log/v3"
@@ -89,11 +87,6 @@ func (c Cfg) WithValuesOnCompressedPage(n int) Cfg {
 	return c
 }
 
-// WorkersLimiter is a caps global active compression workers
-// across all compressor instances in the app. Each worker goroutine acquires 1 from
-// this semaphore before doing CPU-heavy work.
-var WorkersLimiter = semaphore.NewWeighted(int64(dbg.CompressWorkers))
-
 var DefaultCfg = Cfg{
 	MinPatternScore: 1024,
 	MinPatternLen:   5,
@@ -229,8 +222,8 @@ func (c *Compressor) Count() int { return int(c.wordsCount) }
 // 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(128*datasize.KB)) }}
-	bufioReaderPool = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, int(128*datasize.KB)) }}
+	bufioWriterPool = sync.Pool{New: func() any { return bufio.NewWriterSize(nil, int(256*datasize.KB)) }}
+	bufioReaderPool = sync.Pool{New: func() any { return bufio.NewReaderSize(nil, int(256*datasize.KB)) }}
 )
 
 func getBufioWriter(w io.Writer) *bufio.Writer {
@@ -309,11 +302,6 @@ func (c *Compressor) AddUncompressedWord(word []byte) error {
 }
 
 func (c *Compressor) Compress() error {
-	if err := WorkersLimiter.Acquire(c.ctx, 1); err != nil {
-		return err
-	}
-	defer WorkersLimiter.Release(1)
-
 	if err := c.uncompressedFile.Flush(); err != nil {
 		return err
 	}