[experiment] execution/stagedsync: let Normalize read through the tx's own read set

execution/stagedsync/exec3_parallel.go

@@ -2895,7 +2895,17 @@ func (be *blockExecutor) nextResult(ctx context.Context, pe *parallelExecutor, r
 					}
 					// Mirror txtask.go's genesis rules-clobber so empty allocs (AuRa ZeroAddress) survive.
 					emptyRemoval := be.blockNum != 0 && pe.cfg.chainConfig.IsEIP161Enabled(be.blockNum)
-					normWrites, normErr := rawWrites.Normalize(be.versionMap, txVersion.TxIndex, resultIncarnation, stateReader, domainStorageKeys, emptyRemoval, pe.cfg.chainConfig.Aura != nil, txTask.Rules().IsAmsterdam)
+					// Experiment: serve Normalize's fallback reads from what the worker
+					// already recorded, and assert the domain agrees on every one.
+					normReader := state.NewNormalizeReader(txVersion.TxIndex, be.blockIO.ReadSet(txVersion.TxIndex), be.versionMap, stateReader)
+					normWrites, normErr := rawWrites.Normalize(be.versionMap, txVersion.TxIndex, resultIncarnation, normReader, domainStorageKeys, emptyRemoval, pe.cfg.chainConfig.Aura != nil, txTask.Rules().IsAmsterdam)
+					if state.AssertNormalizeReadsEnabled() && normErr == nil && stateReader != nil {
+						domainWrites, domainErr := rawWrites.Normalize(be.versionMap, txVersion.TxIndex, resultIncarnation, stateReader, domainStorageKeys, emptyRemoval, pe.cfg.chainConfig.Aura != nil, txTask.Rules().IsAmsterdam)
+						if domainErr != nil {
+							return nil, fmt.Errorf("[parallel] normalize cross-check: %w", domainErr)
+						}
+						state.AssertNormalizeMatches(domainWrites, normWrites)
+					}
 					if domainKeysErr != nil {
 						return nil, fmt.Errorf("[parallel] iterate storage prefix for block write normalization: %w", domainKeysErr)
 					}
@@ -3170,7 +3180,8 @@ func (be *blockExecutor) nextResult(ctx context.Context, pe *parallelExecutor, r
 				}
 				emptyRemoval := be.blockNum != 0 && pe.cfg.chainConfig.IsEIP161Enabled(be.blockNum)
 				var normErr error
-				finalizeWrites, normErr = writes.Normalize(be.versionMap, finalVersion.TxIndex, finalVersion.Incarnation, reader, domainStorageKeys, emptyRemoval, pe.cfg.chainConfig.Aura != nil, pe.cfg.chainConfig.IsAmsterdam(tt.Header.Time))
+				finalizeNormReader := state.NewNormalizeReader(finalVersion.TxIndex, be.blockIO.ReadSet(finalVersion.TxIndex), be.versionMap, reader)
+				finalizeWrites, normErr = writes.Normalize(be.versionMap, finalVersion.TxIndex, finalVersion.Incarnation, finalizeNormReader, domainStorageKeys, emptyRemoval, pe.cfg.chainConfig.Aura != nil, pe.cfg.chainConfig.IsAmsterdam(tt.Header.Time))
 				if domainKeysErr != nil {
 					return nil, fmt.Errorf("[parallel] finalize iterate storage prefix for block write normalization: %w", domainKeysErr)
 				}

execution/state/checked_reader.go

@@ -0,0 +1,209 @@
+// Copyright 2024 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 state
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/holiman/uint256"
+
+	"github.com/erigontech/erigon/common/dbg"
+	"github.com/erigontech/erigon/execution/types/accounts"
+)
+
+// assertNormalizeReads runs Normalize a second time against the domain reader
+// and asserts the two outputs match. Diagnostic only: it doubles Normalize.
+var assertNormalizeReads = dbg.EnvBool("NORMALIZE_ASSERT_READS", true)
+
+// AssertNormalizeReadsEnabled reports whether the output-level check is on.
+func AssertNormalizeReadsEnabled() bool { return assertNormalizeReads }
+
+// NewNormalizeReader returns the reader Normalize should use: the tx's own
+// recorded reads first, then the versionMap, then the domain.
+func NewNormalizeReader(txIndex int, reads ReadSet, versionMap *VersionMap, domain StateReader) StateReader {
+	if domain == nil { // nil means "no reader"; Normalize guards on it
+		return nil
+	}
+	return NewVersionedStateReader(txIndex, reads, versionMap, domain)
+}
+
+// CheckedStateReader answers from want and asserts that got agrees, panicking
+// on any disagreement. It exists to settle one question empirically: can
+// Normalize take the values a worker already recorded in its read set instead
+// of re-reading the domain on the apply loop? Every divergence is a case where
+// it cannot, and the panic names it.
+//
+// Diagnostic only, and not side-effect free: see NewNormalizeReader.
+type CheckedStateReader struct {
+	want StateReader // read set -> versionMap -> domain
+	got  StateReader // domain
+}
+
+func NewCheckedStateReader(want, got StateReader) *CheckedStateReader {
+	return &CheckedStateReader{want: want, got: got}
+}
+
+func mismatch(op string, addr accounts.Address, detail string) {
+	panic(fmt.Sprintf("checked reader: %s disagrees for %x: %s", op, addr.Value(), detail))
+}
+
+func (r *CheckedStateReader) ReadAccountData(address accounts.Address) (*accounts.Account, error) {
+	want, err := r.want.ReadAccountData(address)
+	if err != nil {
+		mismatch("ReadAccountData", address, fmt.Sprintf("read-set path errored: %v", err))
+	}
+	got, gotErr := r.got.ReadAccountData(address)
+	if gotErr != nil {
+		mismatch("ReadAccountData", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	switch {
+	case want == nil && got == nil:
+	case want == nil || got == nil:
+		mismatch("ReadAccountData", address, fmt.Sprintf("presence differs: readset=%v domain=%v", want != nil, got != nil))
+	case want.Nonce != got.Nonce || !want.Balance.Eq(&got.Balance) ||
+		want.Incarnation != got.Incarnation || want.CodeHash.Value() != got.CodeHash.Value():
+		mismatch("ReadAccountData", address, fmt.Sprintf(
+			"readset={n:%d bal:%s inc:%d ch:%x} domain={n:%d bal:%s inc:%d ch:%x}",
+			want.Nonce, want.Balance.String(), want.Incarnation, want.CodeHash.Value(),
+			got.Nonce, got.Balance.String(), got.Incarnation, got.CodeHash.Value()))
+	}
+	return want, nil
+}
+
+func (r *CheckedStateReader) ReadAccountStorage(address accounts.Address, key accounts.StorageKey) (uint256.Int, bool, error) {
+	want, wantOK, err := r.want.ReadAccountStorage(address, key)
+	if err != nil {
+		mismatch("ReadAccountStorage", address, fmt.Sprintf("read-set path errored: %v", err))
+	}
+	got, gotOK, gotErr := r.got.ReadAccountStorage(address, key)
+	if gotErr != nil {
+		mismatch("ReadAccountStorage", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	// Absent and present-with-zero are the same slot to the no-op filter: it
+	// drops a zero write either way and keeps a non-zero one either way, so
+	// only the effective value has to agree. The read set reports found=true
+	// for a slot the worker read as zero; the domain reports it absent.
+	wantEff, gotEff := want, got
+	if !wantOK {
+		wantEff = uint256.Int{}
+	}
+	if !gotOK {
+		gotEff = uint256.Int{}
+	}
+	if !wantEff.Eq(&gotEff) {
+		mismatch("ReadAccountStorage", address, fmt.Sprintf(
+			"slot %x: readset={%s,found:%v} domain={%s,found:%v}",
+			key.Value(), want.String(), wantOK, got.String(), gotOK))
+	}
+	return want, wantOK, nil
+}
+
+func (r *CheckedStateReader) ReadAccountCode(address accounts.Address) ([]byte, error) {
+	want, err := r.want.ReadAccountCode(address)
+	if err != nil {
+		mismatch("ReadAccountCode", address, fmt.Sprintf("read-set path errored: %v", err))
+	}
+	got, gotErr := r.got.ReadAccountCode(address)
+	if gotErr != nil {
+		mismatch("ReadAccountCode", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	if !bytes.Equal(want, got) {
+		mismatch("ReadAccountCode", address, fmt.Sprintf("len readset=%d domain=%d", len(want), len(got)))
+	}
+	return want, nil
+}
+
+func (r *CheckedStateReader) ReadAccountCodeSize(address accounts.Address) (int, error) {
+	want, err := r.want.ReadAccountCodeSize(address)
+	if err != nil {
+		return 0, err
+	}
+	got, gotErr := r.got.ReadAccountCodeSize(address)
+	if gotErr != nil {
+		mismatch("ReadAccountCodeSize", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	if want != got {
+		mismatch("ReadAccountCodeSize", address, fmt.Sprintf("readset=%d domain=%d", want, got))
+	}
+	return want, nil
+}
+
+func (r *CheckedStateReader) HasStorage(address accounts.Address) (bool, error) {
+	want, err := r.want.HasStorage(address)
+	if err != nil {
+		return false, err
+	}
+	got, gotErr := r.got.HasStorage(address)
+	if gotErr != nil {
+		mismatch("HasStorage", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	if want != got {
+		mismatch("HasStorage", address, fmt.Sprintf("readset=%v domain=%v", want, got))
+	}
+	return want, nil
+}
+
+func (r *CheckedStateReader) ReadAccountIncarnation(address accounts.Address) (uint64, error) {
+	want, err := r.want.ReadAccountIncarnation(address)
+	if err != nil {
+		return 0, err
+	}
+	got, gotErr := r.got.ReadAccountIncarnation(address)
+	if gotErr != nil {
+		mismatch("ReadAccountIncarnation", address, fmt.Sprintf("domain errored (%v) while read set answered", gotErr))
+	}
+	if want != got {
+		mismatch("ReadAccountIncarnation", address, fmt.Sprintf("readset=%d domain=%d", want, got))
+	}
+	return want, nil
+}
+
+func (r *CheckedStateReader) ReadAccountDataForDebug(address accounts.Address) (*accounts.Account, error) {
+	return r.want.ReadAccountDataForDebug(address)
+}
+
+func (r *CheckedStateReader) SetTrace(trace bool, tracePrefix string) {
+	r.want.SetTrace(trace, tracePrefix)
+}
+
+func (r *CheckedStateReader) Trace() bool { return r.want.Trace() }
+
+func (r *CheckedStateReader) TracePrefix() string { return r.want.TracePrefix() }
+
+// AssertNormalizeMatches panics when two Normalize outputs differ. Compared at
+// the output rather than at each read: the apply-side reader fills the block
+// state cache on a miss, so reading it twice to cross-check is a mutation, not
+// an observation, and fails blocks that otherwise pass.
+func AssertNormalizeMatches(domain, readSet *WriteSet) {
+	if domain.Count() != readSet.Count() {
+		panic(fmt.Sprintf("normalize output differs: domain=%d readset=%d writes",
+			domain.Count(), readSet.Count()))
+	}
+	for h := range domain.AllHeaders() {
+		if !readSet.Has(h) {
+			panic(fmt.Sprintf("normalize output differs: read set lacks %x path=%v key=%x",
+				h.Address.Value(), h.Path, h.Key.Value()))
+		}
+	}
+	for h := range readSet.AllHeaders() {
+		if !domain.Has(h) {
+			panic(fmt.Sprintf("normalize output differs: domain lacks %x path=%v key=%x",
+				h.Address.Value(), h.Path, h.Key.Value()))
+		}
+	}
+}