Skip to content
1 change: 0 additions & 1 deletion db/kv/kv_interface.go
Original file line number Diff line number Diff line change
Expand Up @@ -584,7 +584,6 @@ type TemporalMemBatch interface {
GetLatest(domain Domain, key []byte) (v []byte, step Step, ok bool)
GetDiffset(tx RwTx, blockHash common.Hash, blockNumber uint64) ([DomainLen][]DomainEntryDiff, bool, error)
Merge(other TemporalMemBatch) error
ClearRam()
IndexAdd(table InvertedIdx, key []byte, txNum uint64) (err error)
IteratePrefix(domain Domain, prefix []byte, roTx Tx, it func(k []byte, v []byte) (cont bool, err error)) error
HasPrefix(domain Domain, prefix []byte, roTx Tx) ([]byte, []byte, bool, error)
Expand Down
65 changes: 65 additions & 0 deletions db/state/execctx/close_keeps_ram_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
// 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 <http://www.gnu.org/licenses/>.

package execctx_test

import (
"context"
"testing"

"github.com/stretchr/testify/require"

"github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/db/rawdb/rawtemporaldb"
"github.com/erigontech/erigon/db/state/execctx"
)

// TestClose_KeepsDomainRamForReaders pins the lifetime guarantee readers rely
// on: a read view keeps its DomainReader pointing at the SD's in-memory domain
// maps, so Close must release writer resources without clearing those maps —
// otherwise in-flight domain reads (e.g. the head block's receipts) silently
// miss and fall back to a tx that does not have the data yet.
func TestClose_KeepsDomainRamForReaders(t *testing.T) {
t.Parallel()
db := newTestDb(t, 16)
ctx := context.Background()

tx, err := db.BeginTemporalRo(ctx)
require.NoError(t, err)
defer tx.Rollback()

sd, err := execctx.NewSharedDomains(ctx, tx, log.New())
require.NoError(t, err)
require.NoError(t, sd.InitBlockOverlay(tx, t.TempDir()))

const txNum = 7
const cumGas = 21000
require.NoError(t, rawtemporaldb.AppendReceiptMetadata(sd.AsPutDel(tx), 0, cumGas, 0, txNum))

view := sd.BlockOverlayTemporalTx(tx)
require.NotNil(t, view)

assertReceiptVisible := func(msg string) {
got, _, _, err := rawtemporaldb.ReceiptAsOf(view, txNum+1)
require.NoError(t, err)
require.Equal(t, uint64(cumGas), got, msg)
}
assertReceiptVisible("the in-flight receipt must be visible through the view before Close")

sd.Close()

assertReceiptVisible("Close must not clear the domain RAM readers still hold")
}
13 changes: 2 additions & 11 deletions db/state/execctx/domain_shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -889,15 +889,6 @@ func (sd *SharedDomains) PrintCacheStats() {
}
}

func (sd *SharedDomains) ClearRam(resetCommitment bool) {
// When the commitment calculator goroutine owns the Updates buffer,
// skip ClearRam on the commitment context to avoid concurrent btree access.
if resetCommitment && sd.sdCtx != nil && !sd.disableInlineTouchKey {
sd.sdCtx.ClearRam()
}
sd.mem.ClearRam()
}

func (sd *SharedDomains) Size() uint64 {
return sd.mem.SizeEstimate()
}
Expand Down Expand Up @@ -1796,8 +1787,8 @@ func (sd *SharedDomains) DomainDelPrefix(domain kv.Domain, roTx kv.TemporalTx, p
return nil
}

// DiscardWrites disables updates collection for further flushing into db.
// Instead, it keeps them temporarily available until .ClearRam/.Close will make them unavailable.
// DiscardWrites disables updates collection for further flushing into db;
// the values stay readable in memory.
func (sd *SharedDomains) DiscardWrites(d kv.Domain) {
// TODO: Deprecated - need convert this method to Constructor-Builder configuration
if d >= kv.DomainLen {
Expand Down
2 changes: 1 addition & 1 deletion db/state/execctx/statecache_rpc_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ func TestEmbeddedRPCCacheViewDoesNotRefillCodeOfDeletedAccount(t *testing.T) {

got, err := rpcView.GetCode(addr)
require.NoError(t, err)
require.Equal(t, code, got, "the pre-deletion view still reads the code from its own tx")
require.Empty(t, got, "the view keeps serving the published SD's state after teardown, so the deletion stays visible")

_, ok = stateCache.View(nil).Get(kv.CodeDomain, addr)
require.False(t, ok, "a pre-deletion RPC view must not refill the deleted account's code")
Expand Down
29 changes: 2 additions & 27 deletions db/state/temporal_mem_batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -362,32 +362,6 @@ func (sd *TemporalMemBatch) SizeEstimate() uint64 {
return uint64(sd.metrics.CachePutSize)
}

func (sd *TemporalMemBatch) ClearRam() {
sd.latestStateLock.Lock()
defer sd.latestStateLock.Unlock()
for i := range sd.domains {
sd.domains[i] = map[string][]dataWithTxNum{}
}

sd.storage = btree2.NewMap[string, []dataWithTxNum](128)
sd.unwindToTxNum = 0
sd.unwindChangeset = nil
sd.unwindChangesetRaw = nil

sd.metrics.Lock()
defer sd.metrics.Unlock()
sd.metrics.CachePutCount = 0
sd.metrics.CachePutSize = 0
sd.metrics.CachePutKeySize = 0
sd.metrics.CachePutValueSize = 0
for _, dm := range sd.metrics.Domains {
dm.CachePutCount = 0
dm.CachePutSize = 0
dm.CachePutKeySize = 0
dm.CachePutValueSize = 0
}
}

func (sd *TemporalMemBatch) IteratePrefix(domain kv.Domain, prefix []byte, roTx kv.Tx, it func(k []byte, v []byte) (cont bool, err error)) error {
sd.latestStateLock.RLock()
defer sd.latestStateLock.RUnlock()
Expand Down Expand Up @@ -633,6 +607,8 @@ func (sd *TemporalMemBatch) IndexAdd(table kv.InvertedIdx, key []byte, txNum uin
panic(fmt.Errorf("unknown index %s", table))
}

// Close releases writer resources but must not clear the in-memory maps:
// readers of a published SD may still hold them.
func (sd *TemporalMemBatch) Close() {
for _, d := range sd.domainWriters {
if d != nil {
Expand All @@ -650,7 +626,6 @@ func (sd *TemporalMemBatch) Close() {
for _, iiWriter := range sd.pastIIWriters {
iiWriter.close()
}
sd.ClearRam()
}

func (sd *TemporalMemBatch) Merge(o kv.TemporalMemBatch) error {
Expand Down
6 changes: 0 additions & 6 deletions execution/commitment/commitmentdb/commitment_context.go
Original file line number Diff line number Diff line change
Expand Up @@ -278,12 +278,6 @@ func (sdc *SharedDomainsCommitmentContext) Reset() {
}
}

func (sdc *SharedDomainsCommitmentContext) ClearRam() {
sdc.updates.Reset()
sdc.Reset()
sdc.stateReader = nil
}

func (sdc *SharedDomainsCommitmentContext) KeysCount() uint64 {
return sdc.updates.Size()
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -135,10 +135,10 @@ func runFromZeroGenesisAllocPreservedAfterResetReExec(t *testing.T) {
// Mimic `stage_exec --reset`: wipe domain tables and reset stage progress.
require.NoError(t, rawdbreset.ResetExec(ctx, emt.DB))

// Now drive execution the SAME way cmd/integration/commands/stages.go:802
// does — direct SpawnExecuteBlocksStage in a loop, with Flush/ClearRam/
// Commit between iterations. This is the path that fails in CI; the
// engine-API InsertChain path above succeeds.
// Now drive execution the SAME way cmd/integration/commands/stages.go
// does — direct SpawnExecuteBlocksStage in a loop, one rwtx and one
// SharedDomains per batch, each committed with doms.Commit. This is the
// path that fails in CI; the engine-API InsertChain path above succeeds.
require.NoError(t, reExecViaIntegrationPath(t, ctx, emt, gen.TopBlock.NumberU64(), emt.cfg.BatchSize, false /*badBlockHalt*/, logger))

postReExec := checkBalance("after-reset-and-integration-reexec")
Expand Down Expand Up @@ -176,24 +176,18 @@ func setupOfflineExec(emt *ExecModuleTester, batchSize datasize.ByteSize, badBlo
}

// reExecViaIntegrationPath drives execution the way cmd/integration/commands/
// stages.go does: SpawnExecuteBlocksStage one batch per rwtx, committing each
// with doms.Commit + ClearRam and reusing the SharedDomains. This bypasses the
// engine API. doms.Commit (not Flush) is load-bearing: it refreshes the
// aggregator BranchCache to match committed state — Flush leaves it stale and
// corrupts the next batch's trie root.
// stages.go does: SpawnExecuteBlocksStage one batch per rwtx and per
// SharedDomains, each committed with doms.Commit. This bypasses the engine
// API. doms.Commit (not Flush) is load-bearing: it refreshes the aggregator
// BranchCache to match committed state — Flush leaves it stale and corrupts
// the next batch's trie root.
func reExecViaIntegrationPath(t *testing.T, ctx context.Context, emt *ExecModuleTester, toBlock uint64, batchSize datasize.ByteSize, badBlockHalt bool, logger log.Logger) error {
t.Helper()

cfg := setupOfflineExec(emt, batchSize, badBlockHalt)

doms, err := newReusedDomains(ctx, emt, logger)
if err != nil {
return err
}
defer doms.Close()

for {
progress, err := execOneBatch(ctx, emt, doms, cfg, toBlock, logger)
progress, err := execOneBatch(ctx, emt, cfg, toBlock, logger)
if err != nil {
return err
}
Expand All @@ -203,34 +197,23 @@ func reExecViaIntegrationPath(t *testing.T, ctx context.Context, emt *ExecModule
}
}

// newReusedDomains opens a SharedDomains seeded from committed state. The seeding
// tx is rolled back right away: each batch re-seeks commitment under its own tx,
// and the SharedDomains keeps no reference to the tx it was built from.
func newReusedDomains(ctx context.Context, emt *ExecModuleTester, logger log.Logger) (*execctx.SharedDomains, error) {
tx, err := emt.DB.BeginTemporalRo(ctx)
// execOneBatch runs a single batch in its own rwtx and its own SharedDomains
// (a fresh one per call avoids reusing a committed, spent one — same as the
// integration tool). doms.Commit commits the tx and refreshes the BranchCache.
// Returns the Execution stage progress after the batch.
func execOneBatch(ctx context.Context, emt *ExecModuleTester, cfg stagedsync.ExecuteBlockCfg, toBlock uint64, logger log.Logger) (uint64, error) {
tx, err := emt.DB.BeginTemporalRw(ctx)
if err != nil {
return nil, err
return 0, err
}
defer tx.Rollback()

doms, err := execctx.NewSharedDomains(ctx, tx, logger)
if err != nil {
return nil, err
}
doms.SetInMemHistoryReads(false)
return doms, nil
}

// execOneBatch runs a single batch in its own rwtx (begin/rollback-on-error/
// commit), reusing doms. doms.Commit commits the tx and refreshes the BranchCache;
// ClearRam drops the flushed batch so doms is clean for the next call. Returns the
// Execution stage progress after the batch.
func execOneBatch(ctx context.Context, emt *ExecModuleTester, doms *execctx.SharedDomains, cfg stagedsync.ExecuteBlockCfg, toBlock uint64, logger log.Logger) (uint64, error) {
tx, err := emt.DB.BeginTemporalRw(ctx)
if err != nil {
return 0, err
}
defer tx.Rollback()
defer doms.Close()
doms.SetInMemHistoryReads(false)

s, err := emt.Sync.StageState(stages.Execution, tx, true, false)
if err != nil {
Expand All @@ -249,7 +232,6 @@ func execOneBatch(ctx context.Context, emt *ExecModuleTester, doms *execctx.Shar
if err := doms.Commit(ctx, tx); err != nil {
return 0, err
}
doms.ClearRam(true)
return progress, nil
}

Expand Down Expand Up @@ -360,13 +342,24 @@ func TestExec_RestoresCommitmentStateReader(t *testing.T) {

cfg := setupOfflineExec(emt, emt.cfg.BatchSize, false /*badBlockHalt*/)

doms, err := newReusedDomains(ctx, emt, logger)
tx, err := emt.DB.BeginTemporalRw(ctx)
require.NoError(t, err)
defer tx.Rollback()

doms, err := execctx.NewSharedDomains(ctx, tx, logger)
require.NoError(t, err)
defer doms.Close()
doms.SetInMemHistoryReads(false)

readerBefore := doms.GetCommitmentContext().StateReader()
_, err = execOneBatch(ctx, emt, doms, cfg, gen.TopBlock.NumberU64(), logger)

s, err := emt.Sync.StageState(stages.Execution, tx, true, false)
require.NoError(t, err)
err = stagedsync.SpawnExecuteBlocksStage(s, emt.Sync, doms, tx, gen.TopBlock.NumberU64(), ctx, cfg, logger)
if err != nil && !errors.Is(err, &stagedsync.ErrLoopExhausted{}) {
require.NoError(t, err)
}
require.NoError(t, doms.Commit(ctx, tx))

require.Equal(t, readerBefore, doms.GetCommitmentContext().StateReader(),
"exec must restore the commitment state reader it found; leaving the parallel calculator's asOfStateReader installed breaks a later foreground SeekCommitment with in-mem history reads disabled")
Expand Down
2 changes: 1 addition & 1 deletion execution/execmodule/forkchoice.go
Original file line number Diff line number Diff line change
Expand Up @@ -576,7 +576,7 @@ func (e *ExecModule) updateForkChoice(ctx context.Context, originalBlockHash, sa
}
defer commitRwTx.Rollback() // idempotent after a successful Commit
// The committed sd is spent; RunLoop closes it and continues on the
// fresh SD built below (no ClearRam reuse).
// fresh SD built below (no reuse).
if err := sd.Commit(ctx, commitRwTx); err != nil {
return nil, nil, fmt.Errorf("updateForkChoice: flush+commit sd after hasMore: %w", err)
}
Expand Down
Loading