diff --git a/db/kv/kv_interface.go b/db/kv/kv_interface.go index e2836f786f3..bb963047cb0 100644 --- a/db/kv/kv_interface.go +++ b/db/kv/kv_interface.go @@ -535,6 +535,7 @@ type TemporalMemBatch interface { DiscardWrites(domain Domain) Unwind(txNumUnwindTo uint64, changeset *[DomainLen][]DomainEntryDiff) GetAsOf(domain Domain, key []byte, ts uint64) (v []byte, ok bool, err error) + HistorySeek(domain Domain, key []byte, ts uint64) (v []byte, ok bool, err error) SetInMemHistoryReads(v bool) InMemHistoryReads() bool } diff --git a/db/kv/membatchwithdb/memory_mutation.go b/db/kv/membatchwithdb/memory_mutation.go index b5166c64a58..8efb943a996 100644 --- a/db/kv/membatchwithdb/memory_mutation.go +++ b/db/kv/membatchwithdb/memory_mutation.go @@ -37,6 +37,11 @@ import ( var _ kv.TemporalRwTx = &MemoryMutation{} +type DomainReader interface { + GetAsOf(name kv.Domain, k []byte, ts uint64) ([]byte, bool, error) + HistorySeek(name kv.Domain, k []byte, ts uint64) ([]byte, bool, error) +} + type MemoryMutation struct { // mu protects concurrent access to the mutation's maps and backing tx. // Read methods (GetOne, Has) acquire RLock; write methods (Put, Delete, @@ -52,6 +57,7 @@ type MemoryMutation struct { clearedTables map[string]struct{} db kv.TemporalTx statelessCursors map[string]kv.RwCursor + DomainReader DomainReader } // NewMemoryBatch creates a pure Go in-memory batch with no OS-thread affinity. @@ -913,6 +919,15 @@ func (m *MemoryMutation) GetLatest(name kv.Domain, k []byte) (v []byte, step kv. } func (m *MemoryMutation) GetAsOf(name kv.Domain, k []byte, ts uint64) (v []byte, ok bool, err error) { + if m.DomainReader != nil { + val, ok, err := m.DomainReader.GetAsOf(name, k, ts) + if err != nil { + return nil, false, err + } + if ok { + return val, true, nil + } + } if m.db == nil { return nil, false, fmt.Errorf("MemoryMutation: domain read requires backing tx (detached overlay)") } @@ -941,6 +956,15 @@ func (m *MemoryMutation) RangeAsOf(name kv.Domain, fromKey, toKey []byte, ts uin } func (m *MemoryMutation) HistorySeek(name kv.Domain, k []byte, ts uint64) (v []byte, ok bool, err error) { + if m.DomainReader != nil { + val, ok, err := m.DomainReader.HistorySeek(name, k, ts) + if err != nil { + return nil, false, err + } + if ok { + return val, true, nil + } + } if m.db == nil { return nil, false, fmt.Errorf("MemoryMutation: history read requires backing tx (detached overlay)") } @@ -1061,6 +1085,7 @@ func (m *MemoryMutation) newReadViewMut(tx kv.Tx) *MemoryMutation { deletedDups: m.deletedDups, clearedTables: m.clearedTables, db: dbTx, + DomainReader: m.DomainReader, } } @@ -1113,30 +1138,61 @@ func (v *OverlayTemporalReadView) Apply(_ context.Context, f func(tx kv.Tx) erro func (v *OverlayTemporalReadView) GetLatest(name kv.Domain, k []byte) ([]byte, kv.Step, error) { return v.temporalTx.GetLatest(name, k) } + func (v *OverlayTemporalReadView) HasPrefix(name kv.Domain, prefix []byte) ([]byte, []byte, bool, error) { return v.temporalTx.HasPrefix(name, prefix) } + func (v *OverlayTemporalReadView) StepsInFiles(entitySet ...kv.Domain) kv.Step { return v.temporalTx.StepsInFiles(entitySet...) } + func (v *OverlayTemporalReadView) GetAsOf(name kv.Domain, k []byte, ts uint64) ([]byte, bool, error) { + // Check DomainReader independently — this method shadows MemoryMutation.GetAsOf + // and falls through to v.temporalTx (not m.db), so the embedded check never fires. + if v.MemoryMutation != nil && v.MemoryMutation.DomainReader != nil { + val, ok, err := v.MemoryMutation.DomainReader.GetAsOf(name, k, ts) + if err != nil { + return nil, false, err + } + if ok { + return val, true, nil + } + } return v.temporalTx.GetAsOf(name, k, ts) } + func (v *OverlayTemporalReadView) RangeAsOf(name kv.Domain, fromKey, toKey []byte, ts uint64, asc order.By, limit int) (stream.KV, error) { return v.temporalTx.RangeAsOf(name, fromKey, toKey, ts, asc, limit) } + func (v *OverlayTemporalReadView) IndexRange(name kv.InvertedIdx, k []byte, fromTs, toTs int, asc order.By, limit int) (stream.U64, error) { return v.temporalTx.IndexRange(name, k, fromTs, toTs, asc, limit) } + func (v *OverlayTemporalReadView) HistorySeek(name kv.Domain, k []byte, ts uint64) ([]byte, bool, error) { + // Check DomainReader independently — this method shadows MemoryMutation.HistorySeek + // and falls through to v.temporalTx (not m.db), so the embedded check never fires. + if v.MemoryMutation != nil && v.MemoryMutation.DomainReader != nil { + val, ok, err := v.MemoryMutation.DomainReader.HistorySeek(name, k, ts) + if err != nil { + return nil, false, err + } + if ok { + return val, true, nil + } + } return v.temporalTx.HistorySeek(name, k, ts) } + func (v *OverlayTemporalReadView) HistoryRange(name kv.Domain, fromTs, toTs int, asc order.By, limit int) (stream.KV, error) { return v.temporalTx.HistoryRange(name, fromTs, toTs, asc, limit) } + func (v *OverlayTemporalReadView) Debug() kv.TemporalDebugTx { return v.temporalTx.Debug() } + func (v *OverlayTemporalReadView) AggTx() any { return v.temporalTx.AggTx() } diff --git a/db/kv/membatchwithdb/memory_mutation_test.go b/db/kv/membatchwithdb/memory_mutation_test.go index e408bffe2cf..523fcd301d4 100644 --- a/db/kv/membatchwithdb/memory_mutation_test.go +++ b/db/kv/membatchwithdb/memory_mutation_test.go @@ -17,6 +17,7 @@ package membatchwithdb_test import ( + "errors" "fmt" "sync" "testing" @@ -859,3 +860,44 @@ func TestMemoryMutationConcurrentDeleteAndRead(t *testing.T) { wg.Wait() } + +// erroringDomainReader fails every domain read, so a caller that swallows the +// error is indistinguishable from a caller that saw no value at all. +type erroringDomainReader struct{ err error } + +func (r erroringDomainReader) GetAsOf(kv.Domain, []byte, uint64) ([]byte, bool, error) { + return nil, false, r.err +} + +func (r erroringDomainReader) HistorySeek(kv.Domain, []byte, uint64) ([]byte, bool, error) { + return nil, false, r.err +} + +// TestDomainReadErrorsPropagate covers both overlay read views: a DomainReader +// error must reach the caller rather than fall through to the committed tx, +// which would silently answer with stale data. +func TestDomainReadErrorsPropagate(t *testing.T) { + t.Parallel() + + _, rwTx := newTestTx(t) + batch, err := membatchwithdb.NewMemoryBatch(rwTx, "", log.Root()) + require.NoError(t, err) + defer batch.Close() + + wantErr := errors.New("domain reader unavailable") + batch.DomainReader = erroringDomainReader{err: wantErr} + + key := []byte{0x2} + for name, tx := range map[string]kv.TemporalTx{ + "MemoryMutation": batch, + "OverlayTemporalReadView": batch.NewTemporalReadView(rwTx), + } { + t.Run(name, func(t *testing.T) { + _, _, err := tx.GetAsOf(kv.ReceiptDomain, key, 1) + require.ErrorIs(t, err, wantErr, "GetAsOf must propagate the DomainReader error") + + _, _, err = tx.HistorySeek(kv.ReceiptDomain, key, 1) + require.ErrorIs(t, err, wantErr, "HistorySeek must propagate the DomainReader error") + }) + } +} diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index de6deaf4b9b..77c5b9f5e2c 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -568,12 +568,15 @@ func (sd *SharedDomains) InitBlockOverlay(tx kv.TemporalTx, tmpDir string) error if err != nil { return fmt.Errorf("init block overlay: %w", err) } + overlay.DomainReader = sd sd.blockOverlay.Store(overlay) return nil } + func (sd *SharedDomains) GetCommitmentCtx() *commitmentdb.SharedDomainsCommitmentContext { return sd.sdCtx } + func (sd *SharedDomains) Logger() log.Logger { return sd.logger } // SetStateCache sets the state cache for faster lookups. @@ -840,6 +843,10 @@ func (sd *SharedDomains) GetAsOf(domain kv.Domain, key []byte, ts uint64) (v []b return sd.mem.GetAsOf(domain, key, ts) } +func (sd *SharedDomains) HistorySeek(domain kv.Domain, key []byte, ts uint64) (v []byte, ok bool, err error) { + return sd.mem.HistorySeek(domain, key, ts) +} + // DomainPut // Optimizations: // - user can provide `prevVal != nil` - then it will not read prev value from storage diff --git a/db/state/execctx/domain_shared_test.go b/db/state/execctx/domain_shared_test.go index ba52779fcbc..69bbb29bab5 100644 --- a/db/state/execctx/domain_shared_test.go +++ b/db/state/execctx/domain_shared_test.go @@ -39,6 +39,7 @@ import ( "github.com/erigontech/erigon/db/kv/mdbx" "github.com/erigontech/erigon/db/kv/rawdbv3" "github.com/erigontech/erigon/db/kv/temporal" + "github.com/erigontech/erigon/db/rawdb/rawtemporaldb" "github.com/erigontech/erigon/db/state" "github.com/erigontech/erigon/db/state/changeset" "github.com/erigontech/erigon/db/state/execctx" @@ -1740,3 +1741,105 @@ func TestSharedDomain_TouchChangedKeysFromHistory(t *testing.T) { require.Equal(t, expectedRootHash, rootHash) } } + +func TestBlockOverlay_DomainReadsRegression(t *testing.T) { + ctx := context.Background() + stepSize := uint64(10) + db := newTestDb(t, stepSize) + + tx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer tx.Rollback() + + sd, err := execctx.NewSharedDomains(ctx, tx, log.New()) + require.NoError(t, err) + defer sd.Close() + + err = sd.InitBlockOverlay(tx, t.TempDir()) + require.NoError(t, err) + + txNum := uint64(42) + key := []byte("some-test-key") + value := []byte("some-test-value") + + // Put value into a domain (e.g. ReceiptDomain) in sd + err = sd.DomainPut(kv.ReceiptDomain, tx, key, value, txNum, nil) + require.NoError(t, err) + + // --- Production path: overlay.NewReadView returns *MemoryMutation --- + // This is the path exercised by Filters.WithTemporalOverlay and + // Filters.WithOverlay in the RPC layer. + overlay := sd.BlockOverlay() + require.NotNil(t, overlay) + readViewTx := overlay.NewReadView(tx) + require.NotNil(t, readViewTx) + + gotVal, ok, err := readViewTx.GetAsOf(kv.ReceiptDomain, key, txNum+1) + require.NoError(t, err) + require.True(t, ok, "NewReadView (*MemoryMutation) GetAsOf must find in-memory receipt data") + require.Equal(t, value, gotVal) + + gotValHist, ok, err := readViewTx.HistorySeek(kv.ReceiptDomain, key, txNum+1) + require.NoError(t, err) + require.True(t, ok, "NewReadView (*MemoryMutation) HistorySeek must find in-memory receipt data") + require.Equal(t, value, gotValHist) + + // --- Secondary path: overlay.NewTemporalReadView returns *OverlayTemporalReadView --- + overlayTx := sd.BlockOverlayTemporalTx(tx) + require.NotNil(t, overlayTx) + + gotVal2, ok, err := overlayTx.GetAsOf(kv.ReceiptDomain, key, txNum+1) + require.NoError(t, err) + require.True(t, ok, "NewTemporalReadView GetAsOf must find in-memory receipt data") + require.Equal(t, value, gotVal2) + + gotValHist2, ok, err := overlayTx.HistorySeek(kv.ReceiptDomain, key, txNum+1) + require.NoError(t, err) + require.True(t, ok, "NewTemporalReadView HistorySeek must find in-memory receipt data") + require.Equal(t, value, gotValHist2) +} + +// TestReceiptAsOf_InFlightBlockLogIndex pins the read that seeds per-transaction log +// indexes. A block whose commit is in flight has its receipt metadata only in +// SharedDomains, and on a history miss DomainRoTx.GetAsOf falls back to GetLatest — so +// a bare read answers with the last committed block's value. The overlay read view must +// see the in-flight value instead. +func TestReceiptAsOf_InFlightBlockLogIndex(t *testing.T) { + t.Parallel() + + ctx := t.Context() + logger := log.New() + db := newTestDb(t, 10) + + tx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer tx.Rollback() + + const ( + committedTxNum = uint64(5) + committedLogIdx = uint32(7) + inFlightTxNum = uint64(9) + inFlightLogIdx = uint32(3) + ) + + committed, err := execctx.NewSharedDomains(ctx, tx, logger) + require.NoError(t, err) + defer committed.Close() + require.NoError(t, rawtemporaldb.AppendReceipt(committed.AsPutDel(tx), committedLogIdx, 0, 0, committedTxNum)) + require.NoError(t, committed.Flush(ctx, tx)) + committed.Close() + + _, _, stale, err := rawtemporaldb.ReceiptAsOf(tx, inFlightTxNum+1) + require.NoError(t, err) + require.Equal(t, committedLogIdx, stale, "precondition: a bare read must return the stale committed value") + + sd, err := execctx.NewSharedDomains(ctx, tx, logger) + require.NoError(t, err) + defer sd.Close() + require.NoError(t, sd.InitBlockOverlay(tx, t.TempDir())) + require.NoError(t, rawtemporaldb.AppendReceipt(sd.AsPutDel(tx), inFlightLogIdx, 0, 0, inFlightTxNum)) + + _, _, got, err := rawtemporaldb.ReceiptAsOf(sd.BlockOverlay().NewReadView(tx), inFlightTxNum+1) + require.NoError(t, err) + require.Equal(t, inFlightLogIdx, got, "must serve the in-flight block's log index, not the last committed one") +} diff --git a/db/state/temporal_mem_batch.go b/db/state/temporal_mem_batch.go index 15e4c482d73..c685d68b095 100644 --- a/db/state/temporal_mem_batch.go +++ b/db/state/temporal_mem_batch.go @@ -263,7 +263,7 @@ func (sd *TemporalMemBatch) getLatest(domain kv.Domain, key []byte) (v []byte, s } func (sd *TemporalMemBatch) GetAsOf(domain kv.Domain, key []byte, ts uint64) (v []byte, ok bool, err error) { - if !sd.inMemHistoryReads { + if !sd.inMemHistoryReads && domain != kv.ReceiptDomain { return nil, false, errors.New("GetAsOf called on TemporalMemBatch with inMemHistoryReads disabled") } sd.latestStateLock.RLock() @@ -324,6 +324,10 @@ func (sd *TemporalMemBatch) GetAsOf(domain kv.Domain, key []byte, ts uint64) (v return unwoundLatest(domain, keyS) } +func (sd *TemporalMemBatch) HistorySeek(domain kv.Domain, key []byte, ts uint64) (v []byte, ok bool, err error) { + return sd.GetAsOf(domain, key, ts) +} + func (sd *TemporalMemBatch) SizeEstimate() uint64 { sd.latestStateLock.RLock() defer sd.latestStateLock.RUnlock() diff --git a/rpc/jsonrpc/receipts/receipts_generator.go b/rpc/jsonrpc/receipts/receipts_generator.go index e0e1af59363..ce4791e2311 100644 --- a/rpc/jsonrpc/receipts/receipts_generator.go +++ b/rpc/jsonrpc/receipts/receipts_generator.go @@ -211,6 +211,7 @@ type PostStateInfo struct { } func (g *Generator) GetReceipt(ctx context.Context, cfg *chain.Config, tx kv.TemporalTx, header *types.Header, txn types.Transaction, index int, txNum uint64, postState *PostStateInfo) (_ *types.Receipt, err error) { + tx = g.filters.WithTemporalOverlay(tx) blockHash := header.Hash() blockNum := header.Number.Uint64() txnHash := txn.Hash() @@ -450,6 +451,7 @@ func (g *Generator) GetReceipt(ctx context.Context, cfg *chain.Config, tx kv.Tem } func (g *Generator) GetReceipts(ctx context.Context, cfg *chain.Config, tx kv.TemporalTx, block *types.Block, opts eth.ReceiptsOpts) (_ types.Receipts, err error) { + tx = g.filters.WithTemporalOverlay(tx) blockHash := block.Hash() blockNum := block.NumberU64() @@ -680,6 +682,7 @@ func (g *Generator) assertEqualReceipts(fromExecution, fromDB *types.Receipt) { } func (g *Generator) GetReceiptsGasUsed(ctx context.Context, tx kv.TemporalTx, block *types.Block, txNumsReader rawdbv3.TxNumsReader) (types.Receipts, error) { + tx = g.filters.WithTemporalOverlay(tx) if receipts, ok := g.receiptsCache.Get(block.Hash()); ok { return receipts, nil } diff --git a/rpc/jsonrpc/receipts/receipts_generator_overlay_test.go b/rpc/jsonrpc/receipts/receipts_generator_overlay_test.go new file mode 100644 index 00000000000..3211edf019e --- /dev/null +++ b/rpc/jsonrpc/receipts/receipts_generator_overlay_test.go @@ -0,0 +1,81 @@ +// 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 receipts_test + +import ( + "testing" + "time" + + "github.com/holiman/uint256" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/db/rawdb/rawtemporaldb" + "github.com/erigontech/erigon/db/state/execctx" + "github.com/erigontech/erigon/execution/protocol/params" + "github.com/erigontech/erigon/execution/tests/blockgen" + "github.com/erigontech/erigon/execution/types" + "github.com/erigontech/erigon/node/shards" + "github.com/erigontech/erigon/rpc/jsonrpc/receipts" + "github.com/erigontech/erigon/rpc/rpchelper" +) + +// TestGetReceiptLogIndexThroughOverlay pins the wiring that lets GetReceipt see a +// block whose commit is in flight: the log index must be resolved through +// Filters.WithTemporalOverlay, not read from the committed tx. The overlay is +// seeded with a value the committed tx does not hold, so only a routed read can +// produce it. +func TestGetReceiptLogIndexThroughOverlay(t *testing.T) { + signer := types.LatestSignerForChainID(nil) + m := mockWithGenerator(t, 2, func(i int, block *blockgen.BlockGen) { + txn, err := types.SignTx( + types.NewTransaction(block.TxNonce(testAddr), testAddr, uint256.NewInt(1), params.TxGas, nil, nil), + *signer, testKey) + require.NoError(t, err) + block.AddTx(txn) + }) + + tx, err := m.DB.BeginTemporalRw(m.Ctx) + require.NoError(t, err) + defer tx.Rollback() + + const blockNum = uint64(2) + block, err := m.BlockReader.BlockByNumber(m.Ctx, tx, blockNum) + require.NoError(t, err) + require.Len(t, block.Transactions(), 1) + + minTxNum, err := m.BlockReader.TxnumReader().Min(m.Ctx, tx, blockNum) + require.NoError(t, err) + txNum := minTxNum + 1 // txIndex 0, past the block's system tx + + const overlayLogIdx = uint32(41) + + sd, err := execctx.NewSharedDomains(m.Ctx, tx, m.Log) + require.NoError(t, err) + defer sd.Close() + require.NoError(t, sd.InitBlockOverlay(tx, t.TempDir())) + require.NoError(t, rawtemporaldb.AppendReceipt(sd.AsPutDel(tx), overlayLogIdx, 0, 0, txNum)) + + events := shards.NewEvents() + events.PublishOverlay(sd) + ff := rpchelper.New(m.Ctx, rpchelper.DefaultFiltersConfig, nil, nil, nil, func() {}, m.Log, events) + + gen := receipts.NewGenerator(m.Dirs, m.BlockReader, m.Engine, nil, time.Minute, ff) + receipt, err := gen.GetReceipt(m.Ctx, m.ChainConfig, tx, block.HeaderNoCopy(), block.Transactions()[0], 0, txNum, nil) + require.NoError(t, err) + require.Equal(t, overlayLogIdx, receipt.FirstLogIndexWithinBlock, + "GetReceipt must resolve the log index through the block overlay") +}