Merge branch 'main' into anacrolix/pls-faster-execution-tests

Author: anacrolix

db/recsplit/recsplit.go

@@ -173,9 +173,10 @@ type RecSplit struct {
 	built              bool // Flag indicating that the hash function has been built and no more keys can be added
 	logger             log.Logger
 
-	noFsync bool // fsync is enabled by default, but tests can manually disable
-	workers int  // Number of parallel goroutines for Build(); 0 or 1 = sequential
-	timings Timings
+	noFsync            bool // fsync is enabled by default, but tests can manually disable
+	forceCollisionOnce bool // test-only: force one collision on the next Build
+	workers            int  // Number of parallel goroutines for Build(); 0 or 1 = sequential
+	timings            Timings
 
 	progress *background.Progress // If set, tracks 0-100%: add-keys fills 0-50%, build fills 50-100%
 }
@@ -875,6 +876,11 @@ func (rs *RecSplit) Build(ctx context.Context) error {
 	if rs.keysAdded != rs.keyExpectedCount {
 		return fmt.Errorf("rs %s expected keys %d, got %d", rs.fileName, rs.keyExpectedCount, rs.keysAdded)
 	}
+	if rs.forceCollisionOnce {
+		rs.forceCollisionOnce = false
+		rs.collision = true
+		return fmt.Errorf("%w: forced for testing", ErrCollision)
+	}
 	if rs.timings.Enabled {
 		rs.timings.AddTook = time.Since(rs.timings.AddStart) // assume Adding data into compressor complete
 		rs.timings.BuildStart = time.Now()
@@ -1102,6 +1108,12 @@ func (rs *RecSplit) Collision() bool {
 	return rs.collision
 }
 
+// ForceCollisionOnce makes the next Build() fail with a collision error.
+// Test-only: used to exercise collision retry paths.
+func (rs *RecSplit) ForceCollisionOnce() {
+	rs.forceCollisionOnce = true
+}
+
 // bucketResultPool is a package-level sync.Pool for reusing bucketResult instances
 var bucketResultPool = &sync.Pool{
 	New: func() interface{} {

db/state/domain.go

@@ -95,6 +95,9 @@ type Domain struct {
 	_visible *domainVisible
 
 	checker *DependencyIntegrityChecker
+
+	// _testBuildAccessorHook - test-only: called with the recsplit before the build loop in buildHashMapAccessor
+	_testBuildAccessorHook func(rs *recsplit.RecSplit)
 }
 
 type domainVisible struct {
@@ -1109,7 +1112,7 @@ func (d *Domain) buildHashMapAccessor(ctx context.Context, fromStep, toStep kv.S
 		NoFsync:    d.noFsync,
 		//Workers:    d.CompressorCfg.Workers,
 	}
-	return buildHashMapAccessor(ctx, data, idxPath, false, cfg, ps, d.logger)
+	return buildHashMapAccessor(ctx, data, idxPath, false, cfg, ps, d.logger, d._testBuildAccessorHook)
 }
 
 func (d *Domain) missedBtreeAccessors(source []*FilesItem, dl dirListing) (l []*FilesItem) {
@@ -1189,7 +1192,7 @@ func (d *Domain) BuildMissedAccessors(ctx context.Context, g *errgroup.Group, ps
 	}
 }
 
-func buildHashMapAccessor(ctx context.Context, g *seg.Reader, idxPath string, values bool, cfg recsplit.RecSplitArgs, ps *background.ProgressSet, logger log.Logger) (err error) {
+func buildHashMapAccessor(ctx context.Context, g *seg.Reader, idxPath string, values bool, cfg recsplit.RecSplitArgs, ps *background.ProgressSet, logger log.Logger, testHook func(*recsplit.RecSplit)) (err error) {
 	_, fileName := filepath.Split(idxPath)
 	count := g.Count()
 	if !values {
@@ -1214,8 +1217,13 @@ func buildHashMapAccessor(ctx context.Context, g *seg.Reader, idxPath string, va
 		}
 	}()
 
-	var keyPos, valPos uint64
+	if testHook != nil {
+		testHook(rs)
+	}
+
 	for {
+		// Reset positions at the start of each iteration to handle collision retries correctly
+		var keyPos, valPos uint64
 		word := make([]byte, 0, 256)
 		if err := ctx.Err(); err != nil {
 			return err

db/state/domain_test.go

@@ -50,6 +50,7 @@ import (
 	"github.com/erigontech/erigon/db/kv/mdbx"
 	"github.com/erigontech/erigon/db/kv/order"
 	"github.com/erigontech/erigon/db/kv/stream"
+	"github.com/erigontech/erigon/db/recsplit"
 	"github.com/erigontech/erigon/db/seg"
 	"github.com/erigontech/erigon/db/state/changeset"
 	"github.com/erigontech/erigon/db/state/statecfg"
@@ -3103,3 +3104,119 @@ func TestDomain_IntegrateDirtyFilesNilGuard(t *testing.T) {
 	require.NotNil(t, foundAfter, "dirty file for step 0 must still exist after nil StaticFiles")
 	require.NotNil(t, foundAfter.decompressor, "dirty file decompressor must not be overwritten by nil StaticFiles")
 }
+
+// filledDomainWithHashMapAccessor creates a domain configured to use AccessorHashMap
+// (like CommitmentDomain) for testing buildHashMapAccessor code paths.
+func filledDomainWithHashMapAccessor(t *testing.T, logger log.Logger) (kv.RwDB, *Domain, uint64) {
+	t.Helper()
+	dirs := datadir2.New(t.TempDir())
+
+	// Start with AccountsDomain config but switch to HashMap accessor
+	cfg := statecfg.Schema.AccountsDomain
+	cfg.Accessors = statecfg.AccessorHashMap // Use HashMap instead of BTree
+
+	// Set version to V1_0_standart to enable HashMap accessor building
+	cfg.FileVersion = statecfg.DomainVersionTypes{
+		DataKV:       version.V1_0_standart,
+		AccessorBT:   version.V1_0_standart,
+		AccessorKVEI: version.V1_0_standart,
+		AccessorKVI:  version.V1_0_standart,
+	}
+	cfg.Hist.IiCfg.FileVersion = statecfg.IIVersionTypes{
+		DataEF:      version.V1_0_standart,
+		AccessorEFI: version.V1_0_standart,
+	}
+
+	db := mdbx.New(dbcfg.ChainDB, logger).InMem(t, dirs.Chaindata).MustOpen()
+	t.Cleanup(db.Close)
+	salt := uint32(1)
+
+	d, err := NewDomain(cfg, 16, config3.DefaultStepsInFrozenFile, dirs, logger)
+	require.NoError(t, err)
+	d.salt.Store(&salt)
+	d.DisableFsync()
+	t.Cleanup(d.Close)
+
+	txs := fillDomain(t, d, db, logger)
+	return db, d, txs
+}
+
+// collateAndMergeWithCollisionRetry is like collateAndMerge but forces a
+// recsplit collision retry on every buildHashMapAccessor call during merge.
+func collateAndMergeWithCollisionRetry(t *testing.T, tx kv.RwTx, d *Domain, txs uint64) {
+	t.Helper()
+	logEvery := time.NewTicker(30 * time.Second)
+	defer logEvery.Stop()
+	ctx := context.Background()
+
+	// Collate without collision forcing first
+	for step := kv.Step(0); step < kv.Step(txs/d.stepSize)-1; step++ {
+		require.NoError(t, d.collateBuildIntegrate(ctx, step, tx, background.NewProgressSet()))
+	}
+
+	// Now set up collision forcing for merge
+	d._testBuildAccessorHook = func(rs *recsplit.RecSplit) {
+		rs.ForceCollisionOnce()
+	}
+
+	domainRoTx := d.BeginFilesRo()
+	defer domainRoTx.Close()
+
+	// Merge with collision retry
+	r := domainRoTx.findMergeRange(d.dirtyFilesEndTxNumMinimax(), d.dirtyFilesEndTxNumMinimax())
+	if r.values.needMerge {
+		valuesOuts, indexOuts, historyOuts := domainRoTx.staticFilesInRange(r)
+		valuesIn, indexIn, historyIn, err := domainRoTx.mergeFiles(ctx, valuesOuts, indexOuts, historyOuts, r, nil, background.NewProgressSet())
+		require.NoError(t, err)
+		d.integrateMergedDirtyFiles(valuesIn, indexIn, historyIn)
+		d.reCalcVisibleFiles(d.dirtyFilesEndTxNumMinimax())
+	}
+}
+
+// TestDomain_KeyPosResetOnCollisionRetry verifies that keyPos and valPos
+// used in buildHashMapAccessor are reset when the build retries due to a
+// recsplit collision. Without the reset, the .kvi index would contain
+// incorrect offsets on the retry pass.
+func TestDomain_KeyPosResetOnCollisionRetry(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	t.Parallel()
+
+	logger := log.New()
+	db, d, txs := filledDomainWithHashMapAccessor(t, logger)
+
+	ctx := context.Background()
+	tx, err := db.BeginRw(ctx)
+	require.NoError(t, err)
+	defer tx.Rollback()
+
+	// Collate and merge with forced collision retry
+	collateAndMergeWithCollisionRetry(t, tx, d, txs)
+	require.NoError(t, tx.Commit())
+
+	// Verify lookups still work correctly after collision retry
+	roTx, err := db.BeginRo(ctx)
+	require.NoError(t, err)
+	defer roTx.Rollback()
+
+	domainRoTx := d.BeginFilesRo()
+	defer domainRoTx.Close()
+
+	// Check that we can look up keys correctly
+	// If keyPos wasn't reset, the index would have wrong offsets
+	for keyNum := uint64(1); keyNum <= uint64(31); keyNum++ {
+		var k [8]byte
+		binary.BigEndian.PutUint64(k[:], keyNum)
+
+		val, _, found, err := domainRoTx.GetLatest(k[:], roTx)
+		require.NoError(t, err, "key %x", k)
+		require.True(t, found, "key %x should be found", k)
+
+		// Expected value is txs/keyNum
+		var expected [8]byte
+		binary.BigEndian.PutUint64(expected[:], txs/keyNum)
+		require.Equal(t, expected[:], val, "key %x value mismatch", k)
+	}
+}

db/state/history.go

@@ -71,6 +71,9 @@ type History struct {
 	// _visibleFiles - underscore in name means: don't use this field directly, use BeginFilesRo()
 	// underlying array is immutable - means it's ready for zero-copy use
 	_visibleFiles []visibleFile
+
+	// _testBuildVIHook - test-only: called with the recsplit before the build loop in buildVI
+	_testBuildVIHook func(rs *recsplit.RecSplit)
 }
 
 func NewHistory(cfg statecfg.HistCfg, stepSize, stepsInFrozenFile uint64, dirs datadir.Dirs, logger log.Logger) (*History, error) {
@@ -276,6 +279,9 @@ func (h *History) buildVI(ctx context.Context, historyIdxPath string, hist, efHi
 	}
 	defer rs.Close()
 	rs.LogLvl(log.LvlTrace)
+	if h._testBuildVIHook != nil {
+		h._testBuildVIHook(rs)
+	}
 
 	seq := &multiencseq.SequenceReader{}
 	it := &multiencseq.SequenceIterator{}

db/state/history_test.go

@@ -322,6 +322,41 @@ func TestHistoryCollationBuild(t *testing.T) {
 	})
 }
 
+// TestHistoryBuildVI_PageCounterResetOnCollisionRetry verifies that the page
+// counter 'i' used for paged history files is reset when buildVI retries due
+// to a recsplit collision. Without the reset, the .vi index would contain
+// incorrect offsets on the retry pass.
+//
+// The bug only manifests when compressedPageValuesCount > 0 (paged history),
+// which happens during merge (not initial collation). This test forces a
+// collision retry during merge and verifies history lookups remain correct.
+func TestHistoryBuildVI_PageCounterResetOnCollisionRetry(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	t.Parallel()
+
+	logger := log.New()
+
+	test := func(t *testing.T, largeValues bool) {
+		t.Helper()
+		db, h, txs := filledHistory(t, largeValues, logger)
+
+		// Force collision retries during buildVI calls.
+		// Set the hook AFTER collation but BEFORE merge, so only merge's
+		// buildVI calls get the forced collision.
+		collateAndMergeHistoryWithCollisionRetry(t, db, h, txs)
+		checkHistoryHistory(t, h, txs)
+	}
+	t.Run("large_values", func(t *testing.T) {
+		test(t, true)
+	})
+	t.Run("small_values", func(t *testing.T) {
+		test(t, false)
+	})
+}
+
 func TestHistoryAfterPrune(t *testing.T) {
 	logger := log.New()
 	logEvery := time.NewTicker(30 * time.Second)
@@ -1071,6 +1106,65 @@ func collateAndMergeHistory(tb testing.TB, db kv.RwDB, h *History, txs uint64, d
 	require.NoError(err)
 }
 
+// collateAndMergeHistoryWithCollisionRetry is like collateAndMergeHistory
+// but forces a recsplit collision retry on every buildVI call during merge.
+func collateAndMergeHistoryWithCollisionRetry(tb testing.TB, db kv.RwDB, h *History, txs uint64) {
+	tb.Helper()
+	require := require.New(tb)
+
+	logEvery := time.NewTicker(30 * time.Second)
+	defer logEvery.Stop()
+	ctx := context.Background()
+	tx, err := db.BeginRwNosync(ctx)
+	require.NoError(err)
+	defer tx.Rollback()
+
+	// Collate without collision forcing
+	for step := kv.Step(0); step < kv.Step(txs/h.stepSize)-1; step++ {
+		require.NoError(h.collateBuildIntegrate(ctx, step, tx, background.NewProgressSet()))
+
+		hc := h.BeginFilesRo()
+		_, err = hc.Prune(ctx, tx, step.ToTxNum(h.stepSize), (step + 1).ToTxNum(h.stepSize), math.MaxUint64, false, logEvery)
+		hc.Close()
+		require.NoError(err)
+	}
+
+	// Enable collision forcing for merge phase only
+	collisionRetries := 0
+	h._testBuildVIHook = func(rs *recsplit.RecSplit) {
+		rs.ForceCollisionOnce()
+		collisionRetries++
+	}
+
+	var r HistoryRanges
+	maxSpan := h.stepSize * config3.DefaultStepsInFrozenFile
+
+	for {
+		if stop := func() bool {
+			hc := h.BeginFilesRo()
+			defer hc.Close()
+			r = hc.findMergeRange(hc.files.EndTxNum(), maxSpan)
+			if !r.any() {
+				return true
+			}
+			indexOuts, historyOuts, err := hc.staticFilesInRange(r)
+			require.NoError(err)
+			indexIn, historyIn, err := hc.mergeFiles(ctx, indexOuts, historyOuts, r, background.NewProgressSet())
+			require.NoError(err)
+			h.integrateMergedDirtyFiles(indexIn, historyIn)
+			h.reCalcVisibleFiles(h.dirtyFilesEndTxNumMinimax())
+			return false
+		}(); stop {
+			break
+		}
+	}
+
+	require.Greater(collisionRetries, 0, "expected at least one buildVI collision retry during merge")
+
+	err = tx.Commit()
+	require.NoError(err)
+}
+
 func TestHistoryMergeFiles(t *testing.T) {
 	if testing.Short() {
 		t.Skip()

db/state/inverted_index.go

@@ -1174,7 +1174,7 @@ func (ii *InvertedIndex) buildMapAccessor(ctx context.Context, fromStep, toStep
 	// each such non-existing key read `MPH` transforms to random
 	// key read. `LessFalsePositives=true` feature filtering-out such cases (with `1/256=0.3%` false-positives).
 
-	if err := buildHashMapAccessor(ctx, data, idxPath, false, cfg, ps, ii.logger); err != nil {
+	if err := buildHashMapAccessor(ctx, data, idxPath, false, cfg, ps, ii.logger, nil); err != nil {
 		return err
 	}
 	return nil