mempool/evmonlyapp: read nonce/balance through the pending overlay; fetch first-seen accounts outside the store lock

giga/evmonly/executor.go

@@ -52,7 +52,9 @@ type Executor struct {
 	pipelineMu      sync.Mutex
 	pipelineDone    chan struct{}
 	pipelineErr     error
-	pipelineChanges *StateChangeSet
+	pipelineChanges *pendingChanges
+	// Counts commits started, so a reader can tell that a block landed between two of its steps.
+	pipelineGeneration uint64
 	// The first commit that failed, kept so no caller can miss it.
 	pipelineFailure error
 }

giga/evmonly/giga_store.go

@@ -92,7 +92,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar
 		snapshot:     snapshot,
 		missingState: e.missingState,
 	}
-	source = newPendingOverlay(source, pending)
+	source = pending.overlay(source)
 
 	e.blockPhases.SetPhase("execute")
 	result, err := e.executePreparedBlock(ctx, req, source)
@@ -201,12 +201,79 @@ func (e *Executor) AwaitCommits() error {
 
 // pipelinePending returns the changes of a block whose commit has not been waited on yet, or nil
 // when the store is caught up.
-func (e *Executor) pipelinePending() *StateChangeSet {
+func (e *Executor) pipelinePending() *pendingChanges {
 	e.pipelineMu.Lock()
 	defer e.pipelineMu.Unlock()
 	return e.pipelineChanges
 }
 
+// LatestAccount is the balance and nonce of an account after the last block this executor ran.
+type LatestAccount struct {
+	Balance *big.Int
+	Nonce   uint64
+}
+
+// ReadLatestAccount returns addr's balance and nonce after the last block this executor ran,
+// without waiting for that block's commit to land. It reports the first failed commit instead of
+// state that lacks the failed block.
+func (e *Executor) ReadLatestAccount(addr common.Address) (LatestAccount, error) {
+	if e.stateStore == nil {
+		return LatestAccount{}, errMissingStateStore
+	}
+	for {
+		e.pipelineMu.Lock()
+		pending, generation, failure := e.pipelineChanges, e.pipelineGeneration, e.pipelineFailureLocked()
+		e.pipelineMu.Unlock()
+		if failure != nil {
+			return LatestAccount{}, failure
+		}
+		snapshot := e.stateStore.OpenView()
+		if snapshot == nil {
+			return LatestAccount{}, errors.New("giga store returned a nil snapshot")
+		}
+		account, ok := e.readLatestAccount(snapshot, pending, generation, addr)
+		snapshot.Close()
+		if ok {
+			return account, nil
+		}
+	}
+}
+
+// readLatestAccount reads addr through pending laid over snapshot. It reports false when another
+// commit started after generation was read, since the view may then hold a later block's writes and
+// pending would replay older values over them; the caller reads again.
+func (e *Executor) readLatestAccount(snapshot gigatypes.EVMStateView, pending *pendingChanges, generation uint64, addr common.Address) (LatestAccount, bool) {
+	e.pipelineMu.Lock()
+	moved := e.pipelineGeneration != generation
+	e.pipelineMu.Unlock()
+	if moved {
+		return LatestAccount{}, false
+	}
+	reader := pending.overlay(gigaSnapshotStateReader{snapshot: snapshot, missingState: e.missingState})
+	if rowReader, ok := reader.(accountSnapshotReader); ok {
+		if row, ok := rowReader.ReadAccount(addr); ok {
+			balance := row.Balance
+			if balance == nil {
+				balance = new(big.Int)
+			}
+			return LatestAccount{Balance: balance, Nonce: row.Nonce}, true
+		}
+	}
+	return LatestAccount{Balance: reader.GetBalance(addr), Nonce: reader.GetNonce(addr)}, true
+}
+
+// pipelineFailureLocked returns the first failed commit, whether or not a waiter has retired it yet.
+// Callers hold pipelineMu.
+func (e *Executor) pipelineFailureLocked() error {
+	if e.pipelineFailure != nil {
+		return e.pipelineFailure
+	}
+	if e.pipelineErr != nil {
+		return fmt.Errorf("commit state changes: %w", e.pipelineErr)
+	}
+	return nil
+}
+
 // awaitPipelineCommit blocks until the in-flight commit has landed, reporting the first commit that
 // failed. After it returns the store holds every block this executor has run, so the next view
 // opens on a known height and needs no overlay.
@@ -248,14 +315,15 @@ func (e *Executor) awaitPipelineCommit() error {
 // Commits stay ordered because only one is ever in flight: awaitPipelineCommit lands the previous
 // one before this is called.
 func (e *Executor) startPipelineCommit(blockNumber int64, changesets []*proto.NamedChangeSet, changes *StateChangeSet) error {
-	pending := changes.clone()
+	pending := newPendingChanges(changes.clone())
 	done := make(chan struct{})
 	e.pipelineMu.Lock()
 	if failure := e.pipelineFailure; failure != nil {
 		e.pipelineMu.Unlock()
 		return failure
 	}
 	e.pipelineChanges = pending
+	e.pipelineGeneration++
 	e.pipelineDone = done
 	e.pipelineErr = nil
 	e.pipelineMu.Unlock()

giga/evmonly/pipeline_overlay.go

@@ -18,7 +18,12 @@ import (
 // a returned balance or code in place would corrupt the pending state for every other reader.
 type pendingOverlay struct {
 	base StateReader
+	*pendingChanges
+}
 
+// pendingChanges is a StateChangeSet indexed by address and slot, built once per block so every
+// reader that lays it over a view shares the index.
+type pendingChanges struct {
 	balances map[common.Address]*big.Int
 	nonces   map[common.Address]uint64
 	code     map[common.Address][]byte
@@ -28,14 +33,26 @@ type pendingOverlay struct {
 	cleared map[common.Address]struct{}
 }
 
-// newPendingOverlay indexes changes for lookup. It returns base unchanged when there is nothing to
+// newPendingOverlay lays changes over base. It returns base unchanged when there is nothing to
 // overlay, so a caller pays nothing for the first block or after a commit has caught up.
 func newPendingOverlay(base StateReader, changes *StateChangeSet) StateReader {
-	if changes == nil || changes.isEmpty() {
+	return newPendingChanges(changes).overlay(base)
+}
+
+// overlay returns base with the pending changes laid over it, or base itself when there are none.
+func (c *pendingChanges) overlay(base StateReader) StateReader {
+	if c == nil {
 		return base
 	}
-	o := &pendingOverlay{
-		base:     base,
+	return &pendingOverlay{base: base, pendingChanges: c}
+}
+
+// newPendingChanges indexes changes for lookup, or returns nil when they would change nothing.
+func newPendingChanges(changes *StateChangeSet) *pendingChanges {
+	if changes == nil || changes.isEmpty() {
+		return nil
+	}
+	o := &pendingChanges{
 		balances: make(map[common.Address]*big.Int, len(changes.Balances)),
 		nonces:   make(map[common.Address]uint64, len(changes.Nonces)),
 		code:     make(map[common.Address][]byte, len(changes.Code)),

giga/evmonly/pipeline_store_test.go

@@ -169,6 +169,161 @@ func TestRetiringACommitWhileOpeningAViewKeepsThePreviousBlocksState(t *testing.
 	require.Contains(t, second.ChangeSet.Balances, BalanceChange{Address: last, Balance: big.NewInt(1_000)})
 }
 
+// The store's view never advances, so the account state a block produced is only reachable through
+// the pending overlay until AwaitCommits. A latest-account read must report it without settling.
+func TestReadLatestAccountSeesTheBlockWhoseCommitIsInFlight(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	sender := crypto.PubkeyToAddress(key.PublicKey)
+	recipient := testAddress(0xa9)
+
+	snapshot := newMemoryGigaSnapshot(40)
+	snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei))
+	store := &recordingGigaStore{snapshot: snapshot}
+	executor := NewExecutor(Config{}, withTestStores(store, NewMemoryReceiptStore(), noopChangeSetEncoder))
+	defer executor.Close()
+
+	before, err := executor.ReadLatestAccount(sender)
+	require.NoError(t, err)
+	require.Equal(t, LatestAccount{Balance: big.NewInt(testFundedBalanceWei)}, before)
+
+	result := executePipelinedBlock(t, executor, chainID, 41,
+		signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil))
+	require.Equal(t, uint64(1), result.Txs[0].Status)
+
+	got, err := executor.ReadLatestAccount(sender)
+	require.NoError(t, err)
+	require.Equal(t, uint64(1), got.Nonce)
+	require.Equal(t, -1, got.Balance.Cmp(big.NewInt(testFundedBalanceWei)), "gas and value must be deducted")
+	paid, err := executor.ReadLatestAccount(recipient)
+	require.NoError(t, err)
+	require.Equal(t, LatestAccount{Balance: big.NewInt(7)}, paid)
+
+	require.NoError(t, executor.AwaitCommits())
+}
+
+// A commit that failed leaves the store behind the run, so a latest-account read reports the
+// failure rather than state that omits the failed block.
+func TestReadLatestAccountReportsAFailedCommit(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	sender := crypto.PubkeyToAddress(key.PublicKey)
+	recipient := testAddress(0xa9)
+
+	snapshot := newMemoryGigaSnapshot(40)
+	snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei))
+	store := &recordingGigaStore{snapshot: snapshot, commitErr: errTestCommitFailed}
+	executor := NewExecutor(Config{}, withTestStores(store, NewMemoryReceiptStore(), noopChangeSetEncoder))
+	defer executor.Close()
+
+	executePipelinedBlock(t, executor, chainID, 41,
+		signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil))
+	require.ErrorIs(t, executor.AwaitCommits(), errTestCommitFailed)
+
+	_, err = executor.ReadLatestAccount(sender)
+	require.ErrorIs(t, err, errTestCommitFailed)
+}
+
+// A failed commit is reported as soon as the commit has returned, before any waiter has retired it.
+func TestReadLatestAccountReportsAFailedCommitNobodyHasAwaited(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	sender := crypto.PubkeyToAddress(key.PublicKey)
+	recipient := testAddress(0xa9)
+
+	snapshot := newMemoryGigaSnapshot(40)
+	snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei))
+	store := &recordingGigaStore{snapshot: snapshot, commitErr: errTestCommitFailed}
+	executor := NewExecutor(Config{}, withTestStores(store, NewMemoryReceiptStore(), noopChangeSetEncoder))
+	defer executor.Close()
+
+	executePipelinedBlock(t, executor, chainID, 41,
+		signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil))
+	executor.pipelineMu.Lock()
+	done := executor.pipelineDone
+	executor.pipelineMu.Unlock()
+	require.NotNil(t, done)
+	<-done
+
+	_, err = executor.ReadLatestAccount(sender)
+	require.ErrorIs(t, err, errTestCommitFailed)
+}
+
+// A block that lands its commit and starts the next one between a reader's pending read and its
+// view must not leave the reader replaying the older block over the newer state; the read starts
+// over instead.
+func TestReadLatestAccountRestartsWhenABlockLandsUnderIt(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	sender := crypto.PubkeyToAddress(key.PublicKey)
+	recipient := testAddress(0xa9)
+
+	snapshot := newMemoryGigaSnapshot(40)
+	snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei))
+	store := &retiringOnOpenStore{recordingGigaStore: &recordingGigaStore{snapshot: snapshot}}
+	executor := NewExecutor(Config{}, withTestStores(store, NewMemoryReceiptStore(), noopChangeSetEncoder))
+	defer executor.Close()
+
+	executePipelinedBlock(t, executor, chainID, 41,
+		signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil))
+
+	// The reader has block 41 in hand as pending; block 42 lands underneath while its view opens.
+	opens := 0
+	store.retire = func() {
+		opens++
+		if opens == 1 {
+			store.retire = nil
+			executePipelinedBlock(t, executor, chainID, 42,
+				signLegacyTx(t, key, chainID, 1, &recipient, big.NewInt(7), nil))
+		}
+	}
+	got, err := executor.ReadLatestAccount(sender)
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), got.Nonce, "the read must reflect block 42, not replay block 41 over it")
+	require.NoError(t, executor.AwaitCommits())
+}
+
+// A reader whose pending block is retired, and then followed by further blocks that land, between
+// its pending read and its view must not replay that retired block over the newer state, even though
+// nothing is pending any more by the time it looks again.
+func TestReadLatestAccountRestartsWhenItsPendingBlockRetiresUnderIt(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	key, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	sender := crypto.PubkeyToAddress(key.PublicKey)
+	recipient := testAddress(0xa9)
+
+	snapshot := newMemoryGigaSnapshot(40)
+	snapshot.setBalance(sender, big.NewInt(testFundedBalanceWei))
+	store := &retiringOnOpenStore{recordingGigaStore: &recordingGigaStore{snapshot: snapshot}}
+	executor := NewExecutor(Config{}, withTestStores(store, NewMemoryReceiptStore(), noopChangeSetEncoder))
+	defer executor.Close()
+
+	executePipelinedBlock(t, executor, chainID, 41,
+		signLegacyTx(t, key, chainID, 0, &recipient, big.NewInt(7), nil))
+
+	// The reader has block 41 in hand as pending; while its view opens, block 42 runs and both
+	// blocks land, leaving nothing pending and a store view that already holds them.
+	opens := 0
+	store.retire = func() {
+		opens++
+		if opens == 1 {
+			store.retire = nil
+			executePipelinedBlock(t, executor, chainID, 42,
+				signLegacyTx(t, key, chainID, 1, &recipient, big.NewInt(7), nil))
+			require.NoError(t, executor.AwaitCommits())
+			snapshot.nonces[sender] = 2
+		}
+	}
+	got, err := executor.ReadLatestAccount(sender)
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), got.Nonce, "the read must not replay retired block 41 over the landed state")
+}
+
 // An executor without a receipt store commits state only; the block result still carries receipts.
 func TestNoReceiptStoreCommitsStateOnly(t *testing.T) {
 	chainID := big.NewInt(testChainID)

sei-tendermint/internal/evmonlyapp/app.go

@@ -470,20 +470,34 @@ func (a *evmOnlyApplication) callBlockContext() (evmonly.BlockContext, error) {
 	panic("unreachable")
 }
 
-func (a *evmOnlyApplication) EvmNonce(address common.Address) uint64 {
+// latestAccount returns address's balance and nonce after the last finalized block, read through
+// the executor's in-flight commit rather than waiting for it. Before InitChain, or once a commit
+// has failed, it reads the settled store instead.
+func (a *evmOnlyApplication) latestAccount(address common.Address) evmonly.LatestAccount {
+	if executor, ok := a.settler.Load().Get(); ok {
+		if account, err := executor.ReadLatestAccount(address); err == nil {
+			return account
+		}
+	}
 	snapshot := a.openSettledView()
 	defer snapshot.Close()
-	return snapshot.GetNonce(evmOnlyStoreAddress(address))
+	storeAddress := evmOnlyStoreAddress(address)
+	if !snapshot.AccountExists(storeAddress) {
+		return evmonly.LatestAccount{Balance: new(big.Int).Set(evmOnlyBaseBalance)}
+	}
+	balance := snapshot.GetBalance(storeAddress)
+	return evmonly.LatestAccount{
+		Balance: new(big.Int).SetBytes(balance[:]),
+		Nonce:   snapshot.GetNonce(storeAddress),
+	}
+}
+
+func (a *evmOnlyApplication) EvmNonce(address common.Address) uint64 {
+	return a.latestAccount(address).Nonce
 }
 
 func (a *evmOnlyApplication) EvmBalance(address common.Address, _ []byte) uint256.Int {
-	snapshot := a.openSettledView()
-	defer snapshot.Close()
-	if !snapshot.AccountExists(evmOnlyStoreAddress(address)) {
-		return *uint256.MustFromBig(evmOnlyBaseBalance)
-	}
-	balance := snapshot.GetBalance(evmOnlyStoreAddress(address))
-	return *new(uint256.Int).SetBytes(balance[:])
+	return *uint256.MustFromBig(a.latestAccount(address).Balance)
 }
 
 func (a *evmOnlyApplication) EvmChainID() uint64 {