diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index 36c6f0c70e..adbfecc2e8 100644 --- a/giga/evmonly/README.md +++ b/giga/evmonly/README.md @@ -217,21 +217,10 @@ are not enabled, the executor attempts optimistic parallel execution. Initial incarnations are split into execution ranges and run through the shared OCC worker pool against the base state. Worker fan-out is clamped to the amount of available work, so small blocks do not spawn idle workers and can still split -down to one transaction per range. Validation then accepts transactions in -block order, comparing each incarnation's recorded balance, nonce, code, -account, and `(address, slot)` storage reads/writes against writes recorded -between that incarnation's source prefix and its own index. - -Acceptance is the serial barrier, but the work behind it is spread across the -pool: every incarnation's writes are indexed by transaction index up front, a -parallel pass checks the pending run of transactions against that index and -reports the first one the frontier would not accept, the accepted run is folded -into the prefix shard by shard (shards are contiguous address ranges), and the -frontier handles only the reported transaction on the calling goroutine. Each -incarnation records the set of shards it touched, so a worker skips whole -results that hold nothing of its own. The merge emits the changeset one shard at -a time on the pool and concatenates the shards in order, which is canonical -address order; a prefix with few keys is merged on the calling goroutine instead. +down to one transaction per range. Validation then walks transaction order, +comparing each incarnation's recorded balance, nonce, code, account, and +`(address, slot)` storage reads/writes against writes accepted after that +incarnation's source prefix. - transactions with no dependency on newly accepted prior writes are retained and accepted in block order without rerunning diff --git a/giga/evmonly/account_reader_test.go b/giga/evmonly/account_reader_test.go index e10c28a371..c5a4b739a1 100644 --- a/giga/evmonly/account_reader_test.go +++ b/giga/evmonly/account_reader_test.go @@ -1,7 +1,6 @@ package evmonly import ( - "bytes" "math/big" "sync/atomic" "testing" @@ -88,36 +87,48 @@ func TestSnapshotReaderFetchesCodeOnlyWhenTheAccountHasSome(t *testing.T) { require.Equal(t, []byte{0x60, 0x00}, account.Code) } -// The merge compares balance, nonce and code against the same row, and resolves that row once per -// touched account across the pool's workers. -func TestParallelMergeResolvesEveryTouchedAccountOnce(t *testing.T) { +// The merge reads every account it is about to compare against through the pool, then the serial +// comparison finds them already resolved. +func TestPrefetchResolvesEveryTouchedAccountOnce(t *testing.T) { snapshot := newMemoryGigaSnapshot(7) - addrs := make([]common.Address, 0, 512) - for i := range 512 { - addr := common.BigToAddress(new(big.Int).Lsh(big.NewInt(int64(i)+1), 150)) + addrs := make([]common.Address, 0, minPrefetchedAccounts+8) + for i := range minPrefetchedAccounts + 8 { + addr := common.BigToAddress(big.NewInt(int64(i) + 1)) snapshot.setBalance(addr, big.NewInt(int64(i)+1)) - snapshot.nonces[addr] = uint64(i) //nolint:gosec // i is non-negative. addrs = append(addrs, addr) } reading := &accountReadingSnapshot{memoryGigaSnapshot: snapshot} state := newBlockSTMState(gigaSnapshotStateReader{snapshot: reading}) for i, addr := range addrs { - state.shard(addr).balances[addr] = big.NewInt(int64(i) + 100) - state.shard(addr).nonces[addr] = uint64(i) + 1 //nolint:gosec // i is non-negative. + state.balances[addr] = big.NewInt(int64(i) + 100) } - pool := newOCCWorkerPool(4) - defer pool.Close() - - var changes StateChangeSet - require.NoError(t, state.changeSetIntoParallel(t.Context(), pool, &changes)) - - require.Equal(t, int64(len(addrs)), reading.reads.Load(), "each row must be read once") - require.Equal(t, state.ChangeSet(), changes, "the parallel merge must match the serial one") - require.Equal(t, int64(2*len(addrs)), reading.reads.Load(), "the serial merge reads each row once too") - require.Len(t, changes.Balances, len(addrs)) - require.Len(t, changes.Nonces, len(addrs)) - for i := 1; i < len(changes.Balances); i++ { - require.Negative(t, bytes.Compare(changes.Balances[i-1].Address[:], changes.Balances[i].Address[:])) + state.prefetchBaseAccounts(t.Context(), newOCCWorkerPool(4)) + + require.Len(t, state.prefetched, len(addrs)) + for i, addr := range addrs { + require.Equal(t, big.NewInt(int64(i)+1), state.prefetched[addr].Balance) + } + + // The comparison that follows reads the prefetched rows rather than the view. + before := reading.reads.Load() + base := newBaseAccounts(state.source, state.prefetched) + for _, addr := range addrs { + base.balance(addr) } + require.Equal(t, before, reading.reads.Load(), "the merge must not re-read what was prefetched") +} + +// Below the threshold the pool costs more than the reads it saves, so the merge reads them itself. +func TestPrefetchIsSkippedForASmallBlock(t *testing.T) { + snapshot := newMemoryGigaSnapshot(7) + addr := testAddress(0xd4) + snapshot.setBalance(addr, big.NewInt(5)) + reading := &accountReadingSnapshot{memoryGigaSnapshot: snapshot} + + state := newBlockSTMState(gigaSnapshotStateReader{snapshot: reading}) + state.balances[addr] = big.NewInt(6) + state.prefetchBaseAccounts(t.Context(), newOCCWorkerPool(4)) + + require.Nil(t, state.prefetched) } diff --git a/giga/evmonly/cmd/evmonly-loadtest/main_test.go b/giga/evmonly/cmd/evmonly-loadtest/main_test.go index 0f5a002709..64c924516b 100644 --- a/giga/evmonly/cmd/evmonly-loadtest/main_test.go +++ b/giga/evmonly/cmd/evmonly-loadtest/main_test.go @@ -951,10 +951,6 @@ func BenchmarkExecuteTransferBlock(b *testing.B) { name: "same_sender_nonce_chain", args: []string{"--same-sender"}, }, - { - name: "erc20_single_contract", - args: []string{"--workload=" + workloadERC20Transfer}, - }, } for _, tc := range tests { b.Run(tc.name, func(b *testing.B) { @@ -969,7 +965,7 @@ func BenchmarkExecuteTransferBlock(b *testing.B) { require.NoError(b, err) state := newGeneratedState() - workload, err := scenarios.NewWorkload(cfg.workload, scenarioConfig(cfg), state) + workload, err := scenarios.NewTransferWorkload(scenarioConfig(cfg), state) require.NoError(b, err) request, err := workload.BuildBlock(b.Context(), 1) require.NoError(b, err) diff --git a/giga/evmonly/executor.go b/giga/evmonly/executor.go index ca03f6fb1f..574e402c1b 100644 --- a/giga/evmonly/executor.go +++ b/giga/evmonly/executor.go @@ -52,9 +52,7 @@ type Executor struct { pipelineMu sync.Mutex pipelineDone chan struct{} pipelineErr error - pipelineChanges *pendingChanges - // Counts commits started, so a reader can tell that a block landed between two of its steps. - pipelineGeneration uint64 + pipelineChanges *StateChangeSet // The first commit that failed, kept so no caller can miss it. pipelineFailure error } diff --git a/giga/evmonly/executor_test.go b/giga/evmonly/executor_test.go index 87aaac9178..e2a109f633 100644 --- a/giga/evmonly/executor_test.go +++ b/giga/evmonly/executor_test.go @@ -4,7 +4,6 @@ import ( "context" "crypto/ecdsa" "errors" - "math" "math/big" "sync" "testing" @@ -730,64 +729,6 @@ func TestExecutorOCCConflictingTransfersMatchSequential(t *testing.T) { require.Equal(t, big.NewInt(int64(txCount*3)), occState.GetBalance(recipient)) } -// A block large enough for the parallel validation pass and the sharded merge, mixing independent -// transfers, nonce chains from one sender, and a hot recipient, must produce the sequential result -// exactly: transaction results, receipts, cumulative gas and the canonical changeset. -func TestExecutorOCCLargeMixedBlockMatchesSequential(t *testing.T) { - chainID := big.NewInt(testChainID) - hot := testAddress(0xaa) - seqState := NewMemoryState() - occState := NewMemoryState() - var rawTxs [][]byte - sign := func(key *ecdsa.PrivateKey, nonce uint64, to common.Address, value int64) { - rawTxs = append(rawTxs, signLegacyTxWithGasPrice(t, key, chainID, nonce, &to, big.NewInt(value), nil, 100_000, big.NewInt(1))) - } - fund := func(key *ecdsa.PrivateKey) { - sender := crypto.PubkeyToAddress(key.PublicKey) - seqState.SetBalance(sender, big.NewInt(1_000_000_000)) - occState.SetBalance(sender, big.NewInt(1_000_000_000)) - } - for i := range 300 { - key, err := crypto.GenerateKey() - require.NoError(t, err) - fund(key) - switch { - case i%7 == 0: - sign(key, 0, hot, 3) - case i%29 == 0: - for nonce := range 3 { - sign(key, uint64(nonce), common.BigToAddress(big.NewInt(int64(50_000+i))), 5) //nolint:gosec // nonce is non-negative. - } - default: - sign(key, 0, common.BigToAddress(big.NewInt(int64(50_000+i))), 7) - } - } - // A late transaction reads the hot balance that the earlier ones credit, so it is rerun after the - // parallel pass has accepted the run before it. - hotKey, err := crypto.GenerateKey() - require.NoError(t, err) - hotSender := crypto.PubkeyToAddress(hotKey.PublicKey) - seqState.SetBalance(hotSender, big.NewInt(1_000_000_000)) - occState.SetBalance(hotSender, big.NewInt(1_000_000_000)) - seqState.SetBalance(hot, big.NewInt(1)) - occState.SetBalance(hot, big.NewInt(1)) - sign(hotKey, 0, hot, 1) - - req := BlockRequest{Context: blockContext(chainID), Txs: rawTxs} - seqResult, err := NewExecutor(Config{MinGasPrice: big.NewInt(0)}, withTestState(seqState)).ExecuteBlock(t.Context(), req) - require.NoError(t, err) - occResult, err := NewExecutor(Config{MinGasPrice: big.NewInt(0), OCCWorkers: 4}, withTestState(occState)).ExecuteBlock(t.Context(), req) - require.NoError(t, err) - - require.True(t, occResult.OCCStats.Attempted) - require.False(t, occResult.OCCStats.Fallback, occResult.OCCStats.FallbackReason) - require.Greater(t, occResult.OCCStats.RerunCount, uint64(0)) - require.Equal(t, seqResult.GasUsed, occResult.GasUsed) - require.Equal(t, seqResult.Txs, occResult.Txs) - require.Equal(t, seqResult.Receipts, occResult.Receipts) - require.Equal(t, seqResult.ChangeSet, occResult.ChangeSet) -} - func TestExecutorOCCFeePayingTransfersDoNotConflictOnCoinbase(t *testing.T) { chainID := big.NewInt(testChainID) txCount := 4 @@ -2173,7 +2114,7 @@ func TestStateDBGetCodeHashTracksCodelessAccountExistenceReads(t *testing.T) { {kind: stateAccessBalance, address: eoa}: {}, }) validation := occValidationResult{} - accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0, math.MaxInt) + accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0) require.False(t, accepted) require.Equal(t, occFallbackReasonConflict, validation.fallbackReason) } @@ -2285,7 +2226,7 @@ func TestValidateSTMConflictMatrix(t *testing.T) { t.Fatalf("unknown access mode %q", access) } validation := occValidationResult{} - accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0, math.MaxInt) + accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0) return accepted, validation } @@ -2327,7 +2268,7 @@ func TestValidateSTMConflictSourcePrefix(t *testing.T) { readSet: map[stateAccessKey]struct{}{key: {}}, gasLimit: 1, gasUsed: 1, - }, 0, 10, sourcePrefix, math.MaxInt) + }, 0, 10, sourcePrefix) return accepted, validation } cases := []struct { diff --git a/giga/evmonly/giga_store.go b/giga/evmonly/giga_store.go index 5f6dfe44ea..1e6538432f 100644 --- a/giga/evmonly/giga_store.go +++ b/giga/evmonly/giga_store.go @@ -92,7 +92,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar snapshot: snapshot, missingState: e.missingState, } - source = pending.overlay(source) + source = newPendingOverlay(source, pending) e.blockPhases.SetPhase("execute") result, err := e.executePreparedBlock(ctx, req, source) @@ -201,79 +201,12 @@ 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() *pendingChanges { +func (e *Executor) pipelinePending() *StateChangeSet { 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. @@ -315,7 +248,7 @@ 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 := newPendingChanges(changes.clone()) + pending := changes.clone() done := make(chan struct{}) e.pipelineMu.Lock() if failure := e.pipelineFailure; failure != nil { @@ -323,7 +256,6 @@ func (e *Executor) startPipelineCommit(blockNumber int64, changesets []*proto.Na return failure } e.pipelineChanges = pending - e.pipelineGeneration++ e.pipelineDone = done e.pipelineErr = nil e.pipelineMu.Unlock() diff --git a/giga/evmonly/occ.go b/giga/evmonly/occ.go index f3379ea42e..19ce7b6e1e 100644 --- a/giga/evmonly/occ.go +++ b/giga/evmonly/occ.go @@ -8,6 +8,7 @@ import ( "math" "math/big" "sort" + "sync/atomic" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" @@ -26,7 +27,6 @@ type occTxExecution struct { gasUsed uint64 gasLimit uint64 commutativeBalanceDeltas map[common.Address]*big.Int - shards occShardSet incarnation int sourcePrefix int err error @@ -90,9 +90,6 @@ func (e *Executor) executeBlockOCC(ctx context.Context, req PreparedBlock, sourc } e.blockPhases.SetPhase("occ_merge") result, err := e.mergeOCCResults(ctx, results, finalState) - if errors.Is(err, errOCCWorkerPoolClosed) { - return e.executeBlockOCCSequentialFallback(ctx, req, source, validation, occFallbackReasonWorkerPoolClosed) - } if err != nil { return nil, err } @@ -199,7 +196,6 @@ func (r occSpeculativeRunner) executeTaskInto(ctx context.Context, task occExecu } result.incarnation = task.incarnation result.sourcePrefix = task.sourcePrefix - result.shards = result.touchedShards() results[task.txIndex] = result return nil } @@ -303,23 +299,8 @@ func (e *Executor) validateBlockSTM( ) ([]occTxExecution, *blockSTMState, occValidationResult, error) { state := newBlockSTMValidationState(source) validation := occValidationResult{} - if err := state.writes.indexResults(ctx, pool, results); err != nil { - return nil, nil, validation, err - } - var backoff serialBackoff for state.nextToValidate < len(results) { - if backoff.parallelPassDue(state.nextToValidate) { - accepted, err := e.acceptValidatedPrefix(ctx, runner, pool, results, state, &validation) - if err != nil { - return nil, nil, validation, err - } - if state.nextToValidate == len(results) { - break - } - backoff.record(state.nextToValidate, accepted) - } - end := backoff.serialEnd(state.nextToValidate, len(results)) - rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation, end) + rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation) if err != nil { return nil, nil, validation, err } @@ -329,58 +310,18 @@ func (e *Executor) validateBlockSTM( if err := runner.runTasks(ctx, pool, []occExecutionTask{*rerun}, results); err != nil { return nil, nil, validation, err } - // The rerun's writes join the index; the previous incarnation's stay, which can only cost a - // later transaction a rerun it did not need, never miss a conflict. - state.writes.addAllAt(rerun.txIndex, results[rerun.txIndex].writeSet) - state.writes.addCommutativeBalanceDeltasAt(rerun.txIndex, results[rerun.txIndex].commutativeBalanceDeltas) } return results, state.prefix, validation, nil } -// serialBackoff decides how far the serial frontier runs before the next parallel pass. -// -// The frontier alternates between a parallel pass, which accepts every result up to the first one -// that needs attention, and the serial frontier, which handles that one. A block whose transactions -// depend on each other one after another makes every parallel pass accept nothing, so after a pass -// that accepts too little to pay for itself the serial frontier keeps going for a stretch that -// doubles each time it happens again, and resets once a pass accepts enough. -type serialBackoff struct { - until int - stretch int -} - -// parallelPassDue reports whether the frontier at nextToValidate has left the serial stretch. -func (b *serialBackoff) parallelPassDue(nextToValidate int) bool { - return nextToValidate >= b.until -} - -// record sets the serial stretch that follows a parallel pass which accepted accepted results. -func (b *serialBackoff) record(nextToValidate int, accepted int) { - if accepted >= occMinParallelValidation { - b.stretch = 0 - return - } - b.stretch = max(2*b.stretch, occMinParallelValidation) - b.until = nextToValidate + b.stretch -} - -// serialEnd returns the index the serial frontier runs up to, at least one past nextToValidate and -// never past n. -func (b *serialBackoff) serialEnd(nextToValidate int, n int) int { - return min(n, max(b.until, nextToValidate+1)) -} - -// validateBlockSTMFrontier accepts results in block order on the calling goroutine until it reaches -// end or a result that has to be rerun, which it returns as a task. func validateBlockSTMFrontier( ctx context.Context, runner occSpeculativeRunner, results []occTxExecution, state *blockSTMValidationState, validation *occValidationResult, - end int, ) (*occExecutionTask, error) { - for state.nextToValidate < end { + for state.nextToValidate < len(results) { if err := ctx.Err(); err != nil { return nil, err } @@ -421,6 +362,8 @@ func validateBlockSTMFrontier( } state.cumulativeGasUsed += result.gasUsed state.prefix.apply(result) + state.writes.addAllAt(txIndex, result.writeSet) + state.writes.addCommutativeBalanceDeltasAt(txIndex, result.commutativeBalanceDeltas) state.nextToValidate++ } return nil, nil @@ -448,7 +391,7 @@ func needsSTMRerun( } return nextToValidate > sourcePrefix, nil } - return !validateSTMResultAgainstPrefix(validation, writes, result, cumulativeGasUsed, gasLimit, sourcePrefix, txIndex), nil + return !validateSTMResultAgainstPrefix(validation, writes, result, cumulativeGasUsed, gasLimit, sourcePrefix), nil } func newSTMRerunTask( @@ -520,9 +463,6 @@ const ( occFallbackReasonWorkerPoolClosed = "worker_pool_closed" ) -// validateSTMResultAgainstPrefix reports whether the result at txIndex, executed against the prefix -// [0, sourcePrefix), still holds once the writes at [sourcePrefix, txIndex) are accepted, recording -// every conflict it finds. func validateSTMResultAgainstPrefix( validation *occValidationResult, writes *stateAccessIndex, @@ -530,14 +470,13 @@ func validateSTMResultAgainstPrefix( cumulativeGasUsed uint64, gasLimit uint64, sourcePrefix int, - txIndex int, ) bool { if err := stmGasValidationError(validation, result, cumulativeGasUsed, gasLimit); err != nil { return false } conflictsBefore := validation.conflictCount - validation.addConflicts("read", writes, result.readSet, sourcePrefix, txIndex) - validation.addConflicts("write", writes, result.writeSet, sourcePrefix, txIndex) + validation.addConflicts("read", writes, result.readSet, sourcePrefix) + validation.addConflicts("write", writes, result.writeSet, sourcePrefix) if validation.conflictCount == conflictsBefore { return true } @@ -546,28 +485,20 @@ func validateSTMResultAgainstPrefix( } func stmGasValidationError(validation *occValidationResult, result occTxExecution, cumulativeGasUsed uint64, gasLimit uint64) error { - reason, err := stmGasFailure(result, cumulativeGasUsed, gasLimit) - if err != nil { - validation.fallbackReason = reason - } - return err -} - -// stmGasFailure returns the fallback reason and error for a result that does not fit the block gas -// accounting after cumulativeGasUsed, or an empty reason and nil error when it does. -func stmGasFailure(result occTxExecution, cumulativeGasUsed uint64, gasLimit uint64) (string, error) { if result.gasUsed > math.MaxUint64-cumulativeGasUsed { - return occFallbackReasonGasOverflow, errors.New(occFallbackReasonGasOverflow) + validation.fallbackReason = occFallbackReasonGasOverflow + return errors.New(occFallbackReasonGasOverflow) } if cumulativeGasUsed > gasLimit || result.gasLimit > gasLimit-cumulativeGasUsed { - return occFallbackReasonGasLimit, core.ErrGasLimitReached + validation.fallbackReason = occFallbackReasonGasLimit + return core.ErrGasLimitReached } - return "", nil + return nil } -func (r *occValidationResult) addConflicts(access string, writes *stateAccessIndex, set map[stateAccessKey]struct{}, sourcePrefix int, txIndex int) { +func (r *occValidationResult) addConflicts(access string, writes *stateAccessIndex, set map[stateAccessKey]struct{}, sourcePrefix int) { for key := range set { - if !writes.conflictsWithin(key, sourcePrefix, txIndex) { + if !writes.conflictsWithAfter(key, sourcePrefix) { continue } if r.conflicts == nil { @@ -644,7 +575,91 @@ func (k stateAccessKind) String() string { } } +// minPrefetchedAccounts is the point below which resolving rows across the pool costs more in +// waking workers than the serial reads it saves. +const minPrefetchedAccounts = 256 + +// prefetchBaseAccounts resolves, across the worker pool, the account rows the merge will compare +// against, leaving them for ChangeSetInto to find already read. +// +// The merge is the block's largest serial phase and most of it is these reads, one address at a +// time. They are independent and read-only, and the OCC workers already read this view +// concurrently during speculation. +func (s *blockSTMState) prefetchBaseAccounts(ctx context.Context, pool *occWorkerPool) { + if pool == nil { + return + } + reader, ok := s.source.(accountSnapshotReader) + if !ok { + return + } + addrs := s.touchedAccounts() + if len(addrs) < minPrefetchedAccounts { + return + } + + snapshots := make([]accountSnapshot, len(addrs)) + served := make([]bool, len(addrs)) + var next atomic.Int64 + // A failure here only leaves rows unread, which the merge then reads itself. + _ = pool.Run(ctx, len(addrs), func(workerCtx context.Context, _ int, _ int) error { + for { + i := int(next.Add(1)) - 1 + if i >= len(addrs) { + return nil + } + if err := workerCtx.Err(); err != nil { + return err + } + if snapshot, hit := reader.ReadAccount(addrs[i]); hit { + snapshots[i] = snapshot + served[i] = true + } + } + }) + + s.prefetched = make(map[common.Address]accountSnapshot, len(addrs)) + for i, addr := range addrs { + if served[i] { + s.prefetched[addr] = snapshots[i] + } + } +} + +// touchedAccounts returns each address the block wrote a balance, nonce, or code for, once. +func (s *blockSTMState) touchedAccounts() []common.Address { + addrs := make([]common.Address, 0, len(s.balances)+len(s.nonces)+len(s.code)) + seen := make(map[common.Address]struct{}, len(s.balances)+len(s.nonces)+len(s.code)) + for _, set := range []func(func(common.Address)){ + func(yield func(common.Address)) { + for addr := range s.balances { + yield(addr) + } + }, + func(yield func(common.Address)) { + for addr := range s.nonces { + yield(addr) + } + }, + func(yield func(common.Address)) { + for addr := range s.code { + yield(addr) + } + }, + } { + set(func(addr common.Address) { + if _, dup := seen[addr]; dup { + return + } + seen[addr] = struct{}{} + addrs = append(addrs, addr) + }) + } + return addrs +} + func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution, finalState *blockSTMState) (*BlockResult, error) { + finalState.prefetchBaseAccounts(ctx, e.occPool) blockResult, err := e.acquireBlockResult(ctx, len(results)) if err != nil { return nil, err @@ -662,136 +677,99 @@ func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution blockResult.Txs[i] = result.txResult blockResult.Receipts[i] = result.receipt } - if err := finalState.changeSetIntoParallel(ctx, e.occPool, &blockResult.ChangeSet); err != nil { - blockResult.Release() - return nil, err - } + finalState.ChangeSetInto(&blockResult.ChangeSet) return blockResult, nil } -// blockSTMState is the accepted prefix of a block's state, split into address shards so that -// applying results and emitting the changeset can proceed shard by shard on different workers. type blockSTMState struct { - source StateReader - shards [occStateShards]blockSTMShard -} - -// blockSTMShard holds the accepted writes for the addresses of one shard. -type blockSTMShard struct { + source StateReader balances map[common.Address]*big.Int nonces map[common.Address]uint64 code map[common.Address][]byte storageClears map[common.Address]struct{} storage map[storageChangeKey]common.Hash + + // Account rows resolved ahead of the merge by prefetchBaseAccounts, or nil when it did not run. + prefetched map[common.Address]accountSnapshot } func newBlockSTMState(source StateReader) *blockSTMState { if source == nil { source = NewMemoryState() } - state := &blockSTMState{source: source} - for i := range state.shards { - state.shards[i] = blockSTMShard{ - balances: map[common.Address]*big.Int{}, - nonces: map[common.Address]uint64{}, - code: map[common.Address][]byte{}, - storageClears: map[common.Address]struct{}{}, - storage: map[storageChangeKey]common.Hash{}, - } + return &blockSTMState{ + source: source, + balances: map[common.Address]*big.Int{}, + nonces: map[common.Address]uint64{}, + code: map[common.Address][]byte{}, + storageClears: map[common.Address]struct{}{}, + storage: map[storageChangeKey]common.Hash{}, } - return state -} - -func (s *blockSTMState) shard(addr common.Address) *blockSTMShard { - return &s.shards[occShardOf(addr)] } func (s *blockSTMState) GetBalance(addr common.Address) *big.Int { - if balance, ok := s.shard(addr).balances[addr]; ok { + if balance, ok := s.balances[addr]; ok { return cloneBig(balance) } return s.source.GetBalance(addr) } func (s *blockSTMState) GetNonce(addr common.Address) uint64 { - if nonce, ok := s.shard(addr).nonces[addr]; ok { + if nonce, ok := s.nonces[addr]; ok { return nonce } return s.source.GetNonce(addr) } func (s *blockSTMState) GetCode(addr common.Address) []byte { - if code, ok := s.shard(addr).code[addr]; ok { + if code, ok := s.code[addr]; ok { return cloneBytes(code) } return s.source.GetCode(addr) } func (s *blockSTMState) GetState(addr common.Address, key common.Hash) common.Hash { - shard := s.shard(addr) - if value, ok := shard.storage[storageChangeKey{address: addr, key: key}]; ok { + if value, ok := s.storage[storageChangeKey{address: addr, key: key}]; ok { return value } - if _, ok := shard.storageClears[addr]; ok { + if _, ok := s.storageClears[addr]; ok { return common.Hash{} } return s.source.GetState(addr, key) } -// apply folds one accepted result into the prefix. func (s *blockSTMState) apply(result occTxExecution) { - s.applyOwned(result, occAllShards) -} - -// applyOwned folds the parts of one accepted result whose addresses fall in the shards owns -// reports true for. Two callers with disjoint ownership can run concurrently. -func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardSet) { for _, change := range result.changeSet.Balances { - if !owns.has(occShardOf(change.Address)) { - continue - } - shard := s.shard(change.Address) delta := result.commutativeBalanceDeltas[change.Address] _, normalWrite := result.writeSet[stateAccessKey{kind: stateAccessBalance, address: change.Address}] if delta != nil && !normalWrite { balance := cloneBig(s.GetBalance(change.Address)) balance.Add(balance, delta) - shard.balances[change.Address] = balance + s.balances[change.Address] = balance continue } - shard.balances[change.Address] = cloneBig(change.Balance) + s.balances[change.Address] = cloneBig(change.Balance) } for _, change := range result.changeSet.Nonces { - if owns.has(occShardOf(change.Address)) { - s.shard(change.Address).nonces[change.Address] = change.Nonce - } + s.nonces[change.Address] = change.Nonce } for _, change := range result.changeSet.Code { - if !owns.has(occShardOf(change.Address)) { - continue - } if change.Delete { - s.shard(change.Address).code[change.Address] = nil + s.code[change.Address] = nil } else { - s.shard(change.Address).code[change.Address] = cloneBytes(change.Code) + s.code[change.Address] = cloneBytes(change.Code) } } for _, addr := range result.changeSet.StorageClears { - if !owns.has(occShardOf(addr)) { - continue - } - shard := s.shard(addr) - shard.storageClears[addr] = struct{}{} - for key := range shard.storage { + s.storageClears[addr] = struct{}{} + for key := range s.storage { if key.address == addr { - delete(shard.storage, key) + delete(s.storage, key) } } } for _, change := range result.changeSet.Storage { - if owns.has(occShardOf(change.Address)) { - s.shard(change.Address).storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value - } + s.storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value } } @@ -802,15 +780,19 @@ func (s *blockSTMState) ChangeSet() StateChangeSet { } // baseAccounts serves an account's pre-block fields, reading the row once however many fields a -// caller asks for. It is scoped to one shard of one merge and is not safe for concurrent use. +// caller asks for. It is scoped to one merge and is not safe for concurrent use. type baseAccounts struct { source StateReader reader accountSnapshotReader seen map[common.Address]accountSnapshot } -func newBaseAccounts(source StateReader) *baseAccounts { - b := &baseAccounts{source: source, seen: map[common.Address]accountSnapshot{}} +func newBaseAccounts(source StateReader, prefetched map[common.Address]accountSnapshot) *baseAccounts { + seen := prefetched + if seen == nil { + seen = map[common.Address]accountSnapshot{} + } + b := &baseAccounts{source: source, seen: seen} b.reader, _ = source.(accountSnapshotReader) return b } @@ -846,49 +828,40 @@ func (b *baseAccounts) balance(addr common.Address) *big.Int { func (b *baseAccounts) nonce(addr common.Address) uint64 { return b.get(addr).Nonce } func (b *baseAccounts) code(addr common.Address) []byte { return b.get(addr).Code } -// ChangeSetInto writes the block's net state changes, in canonical order, on the calling goroutine. func (s *blockSTMState) ChangeSetInto(changes *StateChangeSet) { changes.resetForReuse() - for i := range s.shards { - s.shards[i].changeSetInto(s.source, changes) - } -} - -// changeSetInto appends the shard's net changes, in canonical order, to changes. Shards partition -// the address space in canonical order, so appending shard by shard yields a canonically ordered -// changeset. -func (h *blockSTMShard) changeSetInto(source StateReader, changes *StateChangeSet) { // The three loops below each compare against the same accounts, and balance, nonce and code hash - // share one row. Reading per field would resolve that row three times per address. - base := newBaseAccounts(source) - balanceAddrs := sortedAddressesFromBigMap(h.balances) + // share one row. Reading per field would resolve that row three times per address, on the one + // goroutine a block's merge runs on. + base := newBaseAccounts(s.source, s.prefetched) + balanceAddrs := sortedAddressesFromBigMap(s.balances) for _, addr := range balanceAddrs { - balance := cloneBig(h.balances[addr]) + balance := cloneBig(s.balances[addr]) if balance.Cmp(base.balance(addr)) == 0 { continue } changes.Balances = append(changes.Balances, BalanceChange{Address: addr, Balance: balance}) } - nonceAddrs := sortedAddressesFromUint64Map(h.nonces) + nonceAddrs := sortedAddressesFromUint64Map(s.nonces) for _, addr := range nonceAddrs { - if h.nonces[addr] == base.nonce(addr) { + if s.nonces[addr] == base.nonce(addr) { continue } - changes.Nonces = append(changes.Nonces, NonceChange{Address: addr, Nonce: h.nonces[addr]}) + changes.Nonces = append(changes.Nonces, NonceChange{Address: addr, Nonce: s.nonces[addr]}) } - codeAddrs := sortedAddressesFromBytesMap(h.code) + codeAddrs := sortedAddressesFromBytesMap(s.code) for _, addr := range codeAddrs { - code := cloneBytes(h.code[addr]) + code := cloneBytes(s.code[addr]) if bytes.Equal(code, base.code(addr)) { continue } changes.Code = append(changes.Code, CodeChange{Address: addr, Code: code, Delete: len(code) == 0}) } - storageClearAddrs := sortedAddressesFromSet(h.storageClears) + storageClearAddrs := sortedAddressesFromSet(s.storageClears) changes.StorageClears = append(changes.StorageClears, storageClearAddrs...) - storageKeys := make([]storageChangeKey, 0, len(h.storage)) - for key := range h.storage { + storageKeys := make([]storageChangeKey, 0, len(s.storage)) + for key := range s.storage { storageKeys = append(storageKeys, key) } sort.Slice(storageKeys, func(i, j int) bool { @@ -898,9 +871,9 @@ func (h *blockSTMShard) changeSetInto(source StateReader, changes *StateChangeSe return bytes.Compare(storageKeys[i].key[:], storageKeys[j].key[:]) < 0 }) for _, key := range storageKeys { - value := h.storage[key] - baseValue := source.GetState(key.address, key.key) - if _, cleared := h.storageClears[key.address]; cleared { + value := s.storage[key] + baseValue := s.source.GetState(key.address, key.key) + if _, cleared := s.storageClears[key.address]; cleared { baseValue = common.Hash{} } if value == baseValue { @@ -915,58 +888,31 @@ func (h *blockSTMShard) changeSetInto(source StateReader, changes *StateChangeSe } } -// stateAccessIndex records, per state key and per address, the range of transaction indexes that -// wrote it. It is split into address shards so that disjoint shards can be filled concurrently. type stateAccessIndex struct { - shards [occStateShards]stateAccessShard -} - -// stateAccessShard indexes the writes to the addresses of one shard. -type stateAccessShard struct { - exact map[stateAccessKey]txIndexSpan - account map[common.Address]txIndexSpan - touched map[common.Address]txIndexSpan - commutativeBalance map[common.Address]txIndexSpan -} - -// txIndexSpan is the lowest and highest transaction index recorded for one key. -type txIndexSpan struct { - first int - last int + exact map[stateAccessKey]int + account map[common.Address]int + touched map[common.Address]int + commutativeBalance map[common.Address]int } func newStateAccessIndex() *stateAccessIndex { - index := &stateAccessIndex{} - for i := range index.shards { - index.shards[i] = stateAccessShard{ - exact: map[stateAccessKey]txIndexSpan{}, - account: map[common.Address]txIndexSpan{}, - touched: map[common.Address]txIndexSpan{}, - commutativeBalance: map[common.Address]txIndexSpan{}, - } + return &stateAccessIndex{ + exact: map[stateAccessKey]int{}, + account: map[common.Address]int{}, + touched: map[common.Address]int{}, + commutativeBalance: map[common.Address]int{}, } - return index } -func (i *stateAccessIndex) shard(addr common.Address) *stateAccessShard { - return &i.shards[occShardOf(addr)] -} - -// conflictsWithin reports whether a write recorded for key would invalidate a read or write of it -// by a transaction that executed against the prefix [0, lo) and sits at index hi, i.e. whether the -// key was written at an index in [lo, hi). Only the first and last write of a key are recorded, so -// the answer can be a false positive when both fall outside the range with a gap across it; it is -// never a false negative, and a false positive costs one rerun, not correctness. -func (i *stateAccessIndex) conflictsWithin(key stateAccessKey, lo int, hi int) bool { - shard := i.shard(key.address) - if writtenWithin(shard.exact, key, lo, hi) { +func (i *stateAccessIndex) conflictsWithAfter(key stateAccessKey, sourcePrefix int) bool { + if i.hasWriteAtOrAfter(i.exact, key, sourcePrefix) { return true } - if writtenWithin(shard.account, key.address, lo, hi) { + if i.hasAddressWriteAtOrAfter(i.account, key.address, sourcePrefix) { return true } if key.kind == stateAccessAccount { - if writtenWithin(shard.touched, key.address, lo, hi) { + if i.hasAddressWriteAtOrAfter(i.touched, key.address, sourcePrefix) { return true } } @@ -976,64 +922,57 @@ func (i *stateAccessIndex) conflictsWithin(key stateAccessKey, lo int, hi int) b if key.kind != stateAccessAccount && key.kind != stateAccessBalance { return false } - return writtenWithin(shard.commutativeBalance, key.address, lo, hi) -} - -// writtenWithin reports whether the span recorded for key may contain an index in [lo, hi). -func writtenWithin[K comparable](writes map[K]txIndexSpan, key K, lo int, hi int) bool { - span, ok := writes[key] - return ok && lo < hi && span.last >= lo && span.first < hi + return i.hasAddressWriteAtOrAfter(i.commutativeBalance, key.address, sourcePrefix) } -// addAll records the set as written by transactions at every index. func (i *stateAccessIndex) addAll(set map[stateAccessKey]struct{}) { - i.addSpan(txIndexSpan{first: 0, last: math.MaxInt}, set, occAllShards) + i.addAllAt(math.MaxInt, set) } -// addAllAt records the set as written by the transaction at txIndex. func (i *stateAccessIndex) addAllAt(txIndex int, set map[stateAccessKey]struct{}) { - i.addSpan(txIndexSpan{first: txIndex, last: txIndex}, set, occAllShards) -} - -func (i *stateAccessIndex) addSpan(span txIndexSpan, set map[stateAccessKey]struct{}, owns occShardSet) { for key := range set { - if !owns.has(occShardOf(key.address)) { - continue - } - shard := i.shard(key.address) - recordSpan(shard.exact, key, span) + i.recordWrite(i.exact, key, txIndex) // Exist/Empty account reads depend on account metadata, not storage slots. if key.kind != stateAccessStorage { - recordSpan(shard.touched, key.address, span) + i.recordAddressWrite(i.touched, key.address, txIndex) } if key.kind == stateAccessAccount { - recordSpan(shard.account, key.address, span) + i.recordAddressWrite(i.account, key.address, txIndex) } } } -// addCommutativeBalanceDeltasAt records the non-zero deltas as balance credits by the transaction -// at txIndex. func (i *stateAccessIndex) addCommutativeBalanceDeltasAt(txIndex int, deltas map[common.Address]*big.Int) { - i.addCommutativeBalanceDeltas(txIndex, deltas, occAllShards) -} - -func (i *stateAccessIndex) addCommutativeBalanceDeltas(txIndex int, deltas map[common.Address]*big.Int, owns occShardSet) { for addr, delta := range deltas { - if delta == nil || delta.Sign() == 0 || !owns.has(occShardOf(addr)) { + if delta == nil || delta.Sign() == 0 { continue } - recordSpan(i.shard(addr).commutativeBalance, addr, txIndexSpan{first: txIndex, last: txIndex}) + i.recordAddressWrite(i.commutativeBalance, addr, txIndex) + } +} + +func (i *stateAccessIndex) hasWriteAtOrAfter(writes map[stateAccessKey]int, key stateAccessKey, sourcePrefix int) bool { + txIndex, ok := writes[key] + return ok && txIndex >= sourcePrefix +} + +func (i *stateAccessIndex) hasAddressWriteAtOrAfter(writes map[common.Address]int, addr common.Address, sourcePrefix int) bool { + txIndex, ok := writes[addr] + return ok && txIndex >= sourcePrefix +} + +func (i *stateAccessIndex) recordWrite(writes map[stateAccessKey]int, key stateAccessKey, txIndex int) { + if existing, ok := writes[key]; ok && existing >= txIndex { + return } + writes[key] = txIndex } -// recordSpan widens the span recorded for key to include span. -func recordSpan[K comparable](writes map[K]txIndexSpan, key K, span txIndexSpan) { - if existing, ok := writes[key]; ok { - span.first = min(span.first, existing.first) - span.last = max(span.last, existing.last) +func (i *stateAccessIndex) recordAddressWrite(writes map[common.Address]int, addr common.Address, txIndex int) { + if existing, ok := writes[addr]; ok && existing >= txIndex { + return } - writes[key] = span + writes[addr] = txIndex } type storageChangeKey struct { diff --git a/giga/evmonly/occ_shards.go b/giga/evmonly/occ_shards.go deleted file mode 100644 index 10ced961d2..0000000000 --- a/giga/evmonly/occ_shards.go +++ /dev/null @@ -1,291 +0,0 @@ -package evmonly - -import ( - "context" - "math" - "sync" - "sync/atomic" - - "github.com/ethereum/go-ethereum/common" -) - -// occStateShards is the number of address shards the accepted prefix and the write index are split -// into. Shards are contiguous address ranges, so concatenating per-shard output in shard order keeps -// it in canonical address order. -const occStateShards = 64 - -// occMinParallelValidation is the fewest results a parallel validation pass is worth running for; -// below it, waking the pool costs more than validating on the calling goroutine. -const occMinParallelValidation = 64 - -// occMinParallelMergeKeys is the fewest accepted keys a parallel merge is worth running for; below -// it, the merge runs on the calling goroutine and appends straight into the pooled changeset. -const occMinParallelMergeKeys = 256 - -// occCancellationCheckInterval is how many items a worker handles between context checks. -const occCancellationCheckInterval = 64 - -// occShardSet is a set of shards, one bit per shard. -type occShardSet uint64 - -var _ = [1]struct{}{}[occStateShards-64] // occShardSet has exactly one bit per shard. - -const occAllShards = occShardSet(math.MaxUint64) - -func occShardOf(addr common.Address) int { - return int(addr[0]) * occStateShards / 256 -} - -func (s occShardSet) has(shard int) bool { return s&(1< from. - state.cumulativeGasUsed = cumulative[stop-from] - state.nextToValidate = stop - return stop - from, nil -} - -// cumulativeGasFrom returns, for each result at or after from, the block gas used before it, plus one -// trailing entry for the gas used after the last. It stops at the first result whose gas would -// overflow the counter and returns that index, or len(results) when none does. -func cumulativeGasFrom(results []occTxExecution, from int, gasUsedBefore uint64) ([]uint64, int) { - cumulative := make([]uint64, 1, len(results)-from+1) - cumulative[0] = gasUsedBefore - for i := from; i < len(results); i++ { - gasUsed := results[i].gasUsed - if gasUsed > math.MaxUint64-cumulative[i-from] { - return cumulative, i - } - cumulative = append(cumulative, cumulative[i-from]+gasUsed) - } - return cumulative, len(results) -} - -// firstUnacceptedResult returns the lowest index in [from, to) whose result the serial frontier -// would not accept given every result before it accepted, or to when it would accept them all. -func firstUnacceptedResult( - ctx context.Context, - pool *occWorkerPool, - results []occTxExecution, - writes *stateAccessIndex, - blockGasLimit uint64, - from int, - to int, - cumulative []uint64, -) (int, error) { - var stop atomic.Int64 - stop.Store(int64(to)) - err := pool.Run(ctx, to-from, func(workerCtx context.Context, workerID int, workers int) error { - for i := from + workerID; i < to; i += workers { - if int64(i) >= stop.Load() { - return nil - } - if (i-from)%occCancellationCheckInterval < workers { - if err := workerCtx.Err(); err != nil { - return err - } - } - if stmFrontierAccepts(results[i], writes, cumulative[i-from], blockGasLimit, i) { - continue - } - lowerStop(&stop, int64(i)) - return nil - } - return nil - }) - return int(stop.Load()), err -} - -func lowerStop(stop *atomic.Int64, index int64) { - for { - current := stop.Load() - if current <= index || stop.CompareAndSwap(current, index) { - return - } - } -} - -// stmFrontierAccepts mirrors the serial frontier's decision for a result at txIndex, without -// recording anything: the frontier records the outcome itself when it reaches the result. -func stmFrontierAccepts(result occTxExecution, writes *stateAccessIndex, cumulativeGasUsed uint64, blockGasLimit uint64, txIndex int) bool { - if result.err != nil { - return false - } - if _, err := stmGasFailure(result, cumulativeGasUsed, blockGasLimit); err != nil { - return false - } - for key := range result.readSet { - if writes.conflictsWithin(key, result.sourcePrefix, txIndex) { - return false - } - } - for key := range result.writeSet { - if writes.conflictsWithin(key, result.sourcePrefix, txIndex) { - return false - } - } - return true -} - -// applyRange folds a run of accepted results into the prefix in block order, each worker applying -// the shards it owns. -func (s *blockSTMState) applyRange(ctx context.Context, pool *occWorkerPool, results []occTxExecution) error { - return pool.Run(ctx, occStateShards, func(workerCtx context.Context, workerID int, workers int) error { - owns := occShardsOwnedBy(workerID, workers) - for txIndex, result := range results { - if txIndex%occCancellationCheckInterval == 0 { - if err := workerCtx.Err(); err != nil { - return err - } - } - if result.shards.intersects(owns) { - s.applyOwned(result, owns) - } - } - return nil - }) -} - -// occChangeSetFragments holds one changeset per shard for a parallel merge to fill. -type occChangeSetFragments [occStateShards]StateChangeSet - -// occFragmentPool recycles fragment arrays so their capacity survives across blocks. -var occFragmentPool = sync.Pool{New: func() any { return new(occChangeSetFragments) }} - -// changeSetIntoParallel writes the block's net state changes, in canonical order, computing each -// shard's part on the pool. The shards' base-state reads are what the merge mostly spends its time -// on, and they are independent of each other. -func (s *blockSTMState) changeSetIntoParallel(ctx context.Context, pool *occWorkerPool, changes *StateChangeSet) error { - if pool == nil || s.keyCount() < occMinParallelMergeKeys { - s.ChangeSetInto(changes) - return ctx.Err() - } - changes.resetForReuse() - fragments := occFragmentPool.Get().(*occChangeSetFragments) - defer occFragmentPool.Put(fragments) - err := pool.Run(ctx, occStateShards, func(workerCtx context.Context, workerID int, workers int) error { - for shard := workerID; shard < occStateShards; shard += workers { - if err := workerCtx.Err(); err != nil { - return err - } - fragments[shard].resetForReuse() - s.shards[shard].changeSetInto(s.source, &fragments[shard]) - } - return nil - }) - if err != nil { - return err - } - for i := range fragments { - fragment := &fragments[i] - changes.Balances = append(changes.Balances, fragment.Balances...) - changes.Nonces = append(changes.Nonces, fragment.Nonces...) - changes.Code = append(changes.Code, fragment.Code...) - changes.StorageClears = append(changes.StorageClears, fragment.StorageClears...) - changes.Storage = append(changes.Storage, fragment.Storage...) - } - return nil -} - -// keyCount returns how many keys the accepted prefix holds across all shards. -func (s *blockSTMState) keyCount() int { - count := 0 - for i := range s.shards { - shard := &s.shards[i] - count += len(shard.balances) + len(shard.nonces) + len(shard.code) + len(shard.storageClears) + len(shard.storage) - } - return count -} diff --git a/giga/evmonly/occ_shards_test.go b/giga/evmonly/occ_shards_test.go deleted file mode 100644 index 5dd9069dac..0000000000 --- a/giga/evmonly/occ_shards_test.go +++ /dev/null @@ -1,190 +0,0 @@ -package evmonly - -import ( - "context" - "math" - "math/big" - "testing" - - "github.com/ethereum/go-ethereum/common" - "github.com/stretchr/testify/require" -) - -func TestOCCShardOfIsMonotonicAndCoversEveryShard(t *testing.T) { - seen := map[int]bool{} - previous := 0 - for first := range 256 { - addr := common.Address{byte(first), 0xff} - shard := occShardOf(addr) - require.GreaterOrEqual(t, shard, previous, "shards must follow address order") - require.Less(t, shard, occStateShards) - seen[shard] = true - previous = shard - } - require.Len(t, seen, occStateShards) -} - -func TestOCCShardsOwnedByPartitionsTheShards(t *testing.T) { - for _, workers := range []int{1, 3, 8, 64, 100} { - var union occShardSet - for workerID := range workers { - owned := occShardsOwnedBy(workerID, workers) - require.False(t, union.intersects(owned), "workers=%d worker %d overlaps another", workers, workerID) - union |= owned - } - require.Equal(t, occAllShards, union, "workers=%d", workers) - } -} - -func TestConflictsWithinHonoursBothBounds(t *testing.T) { - addr := testAddress(0xa1) - key := stateAccessKey{kind: stateAccessBalance, address: addr} - writes := newStateAccessIndex() - writes.addAllAt(5, map[stateAccessKey]struct{}{key: {}}) - - require.False(t, writes.conflictsWithin(key, 0, 5), "a write at the reader's own index is not a conflict") - require.False(t, writes.conflictsWithin(key, 0, 3), "a write above the reader is not a conflict") - require.False(t, writes.conflictsWithin(key, 6, 10), "a write below the source prefix is not a conflict") - require.True(t, writes.conflictsWithin(key, 0, 6)) - require.True(t, writes.conflictsWithin(key, 5, 6)) - require.False(t, writes.conflictsWithin(key, 6, 6), "an empty range holds no writes") - require.False(t, writes.conflictsWithin(stateAccessKey{kind: stateAccessBalance, address: testAddress(0xa2)}, 0, 10)) -} - -func TestConflictsWithinSpanOnlyOverApproximates(t *testing.T) { - addr := testAddress(0xa1) - key := stateAccessKey{kind: stateAccessBalance, address: addr} - writes := newStateAccessIndex() - writes.addAllAt(2, map[stateAccessKey]struct{}{key: {}}) - writes.addAllAt(9, map[stateAccessKey]struct{}{key: {}}) - - require.True(t, writes.conflictsWithin(key, 4, 7), "a gap inside the span is reported as a conflict") - require.False(t, writes.conflictsWithin(key, 0, 2)) - require.False(t, writes.conflictsWithin(key, 10, 20)) -} - -func TestCumulativeGasFromStopsBeforeOverflow(t *testing.T) { - results := []occTxExecution{{gasUsed: 10}, {gasUsed: 20}, {gasUsed: math.MaxUint64}, {gasUsed: 1}} - - cumulative, to := cumulativeGasFrom(results, 1, 100) - require.Equal(t, 2, to) - require.Equal(t, []uint64{100, 120}, cumulative) - - cumulative, to = cumulativeGasFrom(results[:2], 0, 0) - require.Equal(t, 2, to) - require.Equal(t, []uint64{0, 10, 30}, cumulative) -} - -func TestFirstUnacceptedResultReturnsTheLowestRejection(t *testing.T) { - pool := newOCCWorkerPool(8) - defer pool.Close() - const count = 1000 - results := make([]occTxExecution, count) - for i := range results { - results[i] = occTxExecution{gasLimit: 1, gasUsed: 1} - } - cumulative, to := cumulativeGasFrom(results, 0, 0) - require.Equal(t, count, to) - writes := newStateAccessIndex() - - stop, err := firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, count, stop, "every result is acceptable") - - for _, rejected := range []int{999, 640, 333, 1} { - results[rejected].err = errOCCMaxIncarnation - stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, rejected, stop) - } - - results[0].err = errOCCMaxIncarnation - stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, 0, stop, "a rejection at from itself stops the pass before it accepts anything") -} - -func TestSerialBackoffDoublesWhileParallelPassesAcceptTooLittle(t *testing.T) { - const n = 10_000 - var backoff serialBackoff - require.True(t, backoff.parallelPassDue(0)) - require.Equal(t, 1, backoff.serialEnd(0, n), "without a stretch the serial frontier takes one result") - - backoff.record(0, 0) - require.False(t, backoff.parallelPassDue(occMinParallelValidation-1)) - require.True(t, backoff.parallelPassDue(occMinParallelValidation)) - require.Equal(t, occMinParallelValidation, backoff.serialEnd(0, n)) - - next := occMinParallelValidation - backoff.record(next, 1) - require.Equal(t, next+2*occMinParallelValidation, backoff.serialEnd(next, n)) - next = backoff.serialEnd(next, n) - backoff.record(next, 1) - require.Equal(t, next+4*occMinParallelValidation, backoff.serialEnd(next, n)) - require.Equal(t, 100, backoff.serialEnd(next, 100), "the stretch never runs past the block") - - next = backoff.serialEnd(next, n) - backoff.record(next, occMinParallelValidation) - require.True(t, backoff.parallelPassDue(next)) - backoff.record(next, 0) - require.Equal(t, next+occMinParallelValidation, backoff.serialEnd(next, n), "a pass that accepts enough resets the stretch") -} - -func TestFirstUnacceptedResultRejectsGasAndConflicts(t *testing.T) { - pool := newOCCWorkerPool(4) - defer pool.Close() - addr := testAddress(0xb1) - key := stateAccessKey{kind: stateAccessNonce, address: addr} - results := make([]occTxExecution, 200) - for i := range results { - results[i] = occTxExecution{gasLimit: 10, gasUsed: 10} - } - results[150].readSet = map[stateAccessKey]struct{}{key: {}} - writes := newStateAccessIndex() - writes.addAllAt(20, map[stateAccessKey]struct{}{key: {}}) - cumulative, to := cumulativeGasFrom(results, 0, 0) - - stop, err := firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, 150, stop, "a read of a write inside [sourcePrefix, txIndex) is rejected") - - results[150].sourcePrefix = 21 - stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, len(results), stop, "a write below the source prefix is already accounted for") - - stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, 10*100, 0, to, cumulative) - require.NoError(t, err) - require.Equal(t, 100, stop, "the first result over the block gas limit is rejected") -} - -func TestTouchedShardsCoversEveryAddressTheResultChanged(t *testing.T) { - written := common.Address{0x00} - credited := common.Address{0x40} - changed := common.Address{0x80} - cleared := common.Address{0xc0} - result := occTxExecution{ - writeSet: map[stateAccessKey]struct{}{{kind: stateAccessNonce, address: written}: {}}, - commutativeBalanceDeltas: map[common.Address]*big.Int{credited: big.NewInt(1)}, - } - result.changeSet.Storage = append(result.changeSet.Storage, StorageChange{Address: changed}) - result.changeSet.StorageClears = append(result.changeSet.StorageClears, cleared) - - touched := result.touchedShards() - for _, addr := range []common.Address{written, credited, changed, cleared} { - require.True(t, touched.has(occShardOf(addr)), "%s", addr) - } - require.False(t, touched.has(occShardOf(common.Address{0x20}))) -} - -func TestSmallMergeStaysOnTheCallingGoroutine(t *testing.T) { - pool := newOCCWorkerPool(4) - pool.Close() - state := newBlockSTMState(NewMemoryState()) - state.shards[0].nonces[common.Address{0x00}] = 1 - require.Less(t, state.keyCount(), occMinParallelMergeKeys) - - var changes StateChangeSet - require.NoError(t, state.changeSetIntoParallel(context.Background(), pool, &changes), "a closed pool is never touched below the threshold") - require.Equal(t, state.ChangeSet(), changes) -} diff --git a/giga/evmonly/pipeline_overlay.go b/giga/evmonly/pipeline_overlay.go index 947d85ab32..b44fc22e23 100644 --- a/giga/evmonly/pipeline_overlay.go +++ b/giga/evmonly/pipeline_overlay.go @@ -18,12 +18,7 @@ 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 @@ -33,26 +28,14 @@ type pendingChanges struct { cleared map[common.Address]struct{} } -// newPendingOverlay lays changes over base. It returns base unchanged when there is nothing to +// newPendingOverlay indexes changes for lookup. 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 { - 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 - } - 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 + return base } - o := &pendingChanges{ + o := &pendingOverlay{ + base: base, 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)), diff --git a/giga/evmonly/pipeline_store_test.go b/giga/evmonly/pipeline_store_test.go index e0802ab80e..733375e573 100644 --- a/giga/evmonly/pipeline_store_test.go +++ b/giga/evmonly/pipeline_store_test.go @@ -169,161 +169,6 @@ 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) diff --git a/sei-tendermint/internal/evmonlyapp/app.go b/sei-tendermint/internal/evmonlyapp/app.go index 324b99793e..7000bfa5cb 100644 --- a/sei-tendermint/internal/evmonlyapp/app.go +++ b/sei-tendermint/internal/evmonlyapp/app.go @@ -12,12 +12,9 @@ import ( "runtime" "slices" "sync" - "sync/atomic" - - "go.opentelemetry.io/otel/attribute" - otelmetric "go.opentelemetry.io/otel/metric" "github.com/ethereum/go-ethereum/common" + "go.opentelemetry.io/otel" ethcore "github.com/ethereum/go-ethereum/core" ethtypes "github.com/ethereum/go-ethereum/core/types" @@ -25,8 +22,6 @@ import ( "github.com/ethereum/go-ethereum/params" "github.com/holiman/uint256" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" - "github.com/sei-protocol/seilog" - "go.opentelemetry.io/otel" "github.com/sei-protocol/sei-chain/giga/evmonly" "github.com/sei-protocol/sei-chain/sei-db/bootstrap" @@ -39,8 +34,6 @@ import ( const evmOnlyMinGasPrice = 1_000_000_000 -var logger = seilog.NewLogger("tendermint", "internal", "evmonlyapp") - // evmOnlyBaseFee is the base fee this application executes every block at. // Admission and block validity both price against it, so they cannot diverge. func evmOnlyBaseFee() *big.Int { return new(big.Int) } @@ -63,14 +56,7 @@ type evmOnlyApplication struct { storage *bootstrap.GigaStorageManager changeSetEncoder evmonly.NamedChangeSetEncoder validators []abci.ValidatorUpdate - // executor is held for the whole of a block's execution, so it serializes - // FinalizeBlock and InitChain against each other. EvmCall only takes it to - // read the executor out; the call itself runs unlocked. - executor utils.Mutex[*utils.Option[*evmonly.Executor]] - // settler publishes the same executor to readers of committed state that - // must not wait for a block to finish executing; they settle its - // background commit before opening a store view. - settler utils.AtomicSend[utils.Option[*evmonly.Executor]] + executor utils.Mutex[*utils.Option[*evmonly.Executor]] // Lock order: executor before cursor. FinalizeBlock holds executor while // the block's cursor encoder takes cursor. cursor utils.Mutex[*evmOnlyCursorState] @@ -78,45 +64,16 @@ type evmOnlyApplication struct { // in CheckTx to the sender recovered there, so execution does not recover // it again. checkedSenders utils.Mutex[map[common.Hash]common.Address] - // settleFailureLogged is set once a failed commit has been logged by a - // committed-state reader; the failure is latched, so it is logged once. - settleFailureLogged atomic.Bool // finalizePhases breaks FinalizeBlock into its stages around the executor. // FinalizeBlock is serialized by executor, so one timer serves the app; it // is only touched with that lock held. finalizePhases *seidbmetrics.PhaseTimer - // prepared holds the block PrepareBlock decoded ahead of FinalizeBlock, if any. - prepared utils.Mutex[*utils.Option[preparedBlock]] - // preparedBlocks counts finalized blocks by whether prepared held them. - preparedBlocks otelmetric.Int64Counter - // preparePhases times PrepareBlock's decode of the next block. PrepareBlock is - // called from the single block fetcher, so one timer serves the app. - preparePhases *seidbmetrics.PhaseTimer -} - -// preparedBlock is the stateless part of a FinalizeBlock request, computed before the -// request arrives. FinalizeBlock uses it only for the block with this height and hash. -type preparedBlock struct { - height int64 - hash common.Hash - txs []evmonly.PreparedTx } // finalizeMeterName is the OTel meter FinalizeBlock's phase timer records to, // as evmonly_finalize_phase_duration_seconds_total. const finalizeMeterName = "evmonly_app" -func newPreparedBlocksCounter(meter otelmetric.Meter) otelmetric.Int64Counter { - counter, err := meter.Int64Counter( - "evmonly_finalize_prepared_blocks_total", - otelmetric.WithDescription("Finalized blocks by whether PrepareBlock had already decoded them"), - ) - if err != nil { - panic(fmt.Sprintf("evmonly_finalize_prepared_blocks_total: %v", err)) - } - return counter -} - // evmOnlyCursorState is the execution position: the block whose state is // committed to storage and the block finalized but not yet acknowledged by // Commit. @@ -153,10 +110,6 @@ func NewEVMOnlyApplication( validators: slices.Clone(validators), executor: utils.NewMutex(new(utils.Option[*evmonly.Executor])), finalizePhases: seidbmetrics.NewPhaseTimer(otel.Meter(finalizeMeterName), "evmonly_finalize"), - prepared: utils.NewMutex(new(utils.Option[preparedBlock])), - preparedBlocks: newPreparedBlocksCounter(otel.Meter(finalizeMeterName)), - preparePhases: seidbmetrics.NewPhaseTimer(otel.Meter(finalizeMeterName), "evmonly_prepare"), - settler: utils.NewAtomicSend(utils.None[*evmonly.Executor]()), cursor: utils.NewMutex(&evmOnlyCursorState{}), checkedSenders: utils.NewMutex(map[common.Hash]common.Address{}), } @@ -166,7 +119,7 @@ func NewEVMOnlyApplication( } if cursor, ok := cursor.Get(); ok { for executor := range a.executor.Lock() { - a.installExecutor(executor) + *executor = utils.Some(a.newExecutor()) } for state := range a.cursor.Lock() { state.committed = cursor @@ -175,26 +128,6 @@ func NewEVMOnlyApplication( return a, nil } -// installExecutor creates the executor and publishes it to both the block -// serializer and the settler. Called with the executor lock held. -func (a *evmOnlyApplication) installExecutor(slot *utils.Option[*evmonly.Executor]) { - executor := a.newExecutor() - *slot = utils.Some(executor) - a.settler.Store(utils.Some(executor)) -} - -// AwaitCommits blocks until every block finalized so far is in the store and -// reports the first commit that failed. The store must be settled before it is -// closed, and before a reader opens a view that has to include the last -// finalized block. -func (a *evmOnlyApplication) AwaitCommits() error { - executor, ok := a.settler.Load().Get() - if !ok { - return nil - } - return executor.AwaitCommits() -} - func (a *evmOnlyApplication) newExecutor() *evmonly.Executor { return evmonly.NewExecutor(evmonly.Config{ ChainConfig: a.chainConfig, @@ -257,7 +190,7 @@ func (a *evmOnlyApplication) InitChain(req *abci.RequestInitChain) (*abci.Respon if err := a.seedInitialStateVersion(req.InitialHeight); err != nil { return nil, err } - a.installExecutor(executor) + *executor = utils.Some(a.newExecutor()) for state := range a.cursor.Lock() { state.committed = evmOnlyCursor{height: req.InitialHeight - 1, gasLimit: gasLimit} } @@ -383,20 +316,6 @@ func (a *evmOnlyApplication) rememberSender(hash common.Hash, sender common.Addr // raw transaction is the keccak of its bytes for every transaction type, so no // decoding is needed. func (a *evmOnlyApplication) takeSenders(txs [][]byte) []utils.Option[common.Address] { - return a.checkedSendersOf(txs, true) -} - -// peekSenders is takeSenders without forgetting the entries. -func (a *evmOnlyApplication) peekSenders(txs [][]byte) []utils.Option[common.Address] { - return a.checkedSendersOf(txs, false) -} - -// forgetSenders drops the CheckTx-recovered senders of txs. -func (a *evmOnlyApplication) forgetSenders(txs [][]byte) { - a.checkedSendersOf(txs, true) -} - -func (a *evmOnlyApplication) checkedSendersOf(txs [][]byte, forget bool) []utils.Option[common.Address] { out := make([]utils.Option[common.Address], len(txs)) // Hashed outside the lock; CheckTx writes this map constantly. hashes := hashRawTxs(txs) @@ -404,9 +323,7 @@ func (a *evmOnlyApplication) checkedSendersOf(txs [][]byte, forget bool) []utils for i, hash := range hashes { if sender, ok := senders[hash]; ok { out[i] = utils.Some(sender) - if forget { - delete(senders, hash) - } + delete(senders, hash) } } } @@ -479,73 +396,20 @@ func evmOnlyStoreAddress(address common.Address) gigatypes.Address { return storeAddress } -// openSettledView opens a store view that holds every block finalized so far. -// A failed commit is logged once rather than returned: the view is still a -// consistent version, and the failure halts the node through the next -// FinalizeBlock. -func (a *evmOnlyApplication) openSettledView() gigatypes.StateView { - if err := a.AwaitCommits(); err != nil && !a.settleFailureLogged.Swap(true) { - logger.Error("EVM-only committed state is behind a failed block commit", "err", err) - } - return a.storage.StateDB().OpenView() -} - -// callBlockContext returns the block context of the committed block, and -// refuses while a finalized block awaits Commit. -func (a *evmOnlyApplication) callBlockContext() (evmonly.BlockContext, error) { - for state := range a.cursor.Lock() { - if state.pending.IsPresent() { - // The store already has this block's writes; NUMBER/TIMESTAMP/PrevRandao advance only on Commit. - return evmonly.BlockContext{}, fmt.Errorf("EVM-only call attempted before committing the finalized block") - } - number, ok := utils.SafeCast[uint64](state.committed.height) - if !ok { - return evmonly.BlockContext{}, fmt.Errorf("EVM-only committed height exceeds uint64: %d", state.committed.height) - } - // Coinbase and ParentHash are left zero: no coinbase is tracked outside - // FinalizeBlock, and only the current block's hash is tracked at all. - return evmonly.BlockContext{ - Number: number, - Time: state.lastBlockTime, - GasLimit: state.committed.gasLimit, - ChainID: new(big.Int).Set(a.chainID), - BaseFee: evmOnlyBaseFee(), - BlobBaseFee: new(big.Int), - BlockHash: state.committed.blockHash, - PrevRandao: state.committed.prevRandao, - }, nil - } - panic("unreachable") -} - -// 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() - 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 + snapshot := a.storage.StateDB().OpenView() + defer snapshot.Close() + return snapshot.GetNonce(evmOnlyStoreAddress(address)) } func (a *evmOnlyApplication) EvmBalance(address common.Address, _ []byte) uint256.Int { - return *uint256.MustFromBig(a.latestAccount(address).Balance) + snapshot := a.storage.StateDB().OpenView() + defer snapshot.Close() + if !snapshot.AccountExists(evmOnlyStoreAddress(address)) { + return *uint256.MustFromBig(evmOnlyBaseBalance) + } + balance := snapshot.GetBalance(evmOnlyStoreAddress(address)) + return *new(uint256.Int).SetBytes(balance[:]) } func (a *evmOnlyApplication) EvmChainID() uint64 { @@ -579,131 +443,52 @@ func (a *evmOnlyApplication) EvmCall(ctx context.Context, msg *ethcore.Message) } executor = got } - // The committed block's write may still be in flight, and a block may be - // finalized and committed while it is waited for. The context is taken - // before settling and confirmed unchanged after, so the store holds the - // advertised block and no later one has been committed to the cursor. - for { - if err := ctx.Err(); err != nil { - return nil, err - } - blockCtx, err := a.callBlockContext() - if err != nil { - return nil, err - } - if err := executor.AwaitCommits(); err != nil { - return nil, err + var blockCtx evmonly.BlockContext + for state := range a.cursor.Lock() { + if state.pending.IsPresent() { + // The store already has this block's writes; NUMBER/TIMESTAMP/PrevRandao advance only on Commit. + return nil, fmt.Errorf("EVM-only call attempted before committing the finalized block") } - settled, err := a.callBlockContext() - if err != nil { - return nil, err + number, ok := utils.SafeCast[uint64](state.committed.height) + if !ok { + return nil, fmt.Errorf("EVM-only committed height exceeds uint64: %d", state.committed.height) } - if settled.Number == blockCtx.Number { - return executor.Call(ctx, blockCtx, msg) + // Coinbase and ParentHash are left zero: no coinbase is tracked outside + // FinalizeBlock, and only the current block's hash is tracked at all. + blockCtx = evmonly.BlockContext{ + Number: number, + Time: state.lastBlockTime, + GasLimit: state.committed.gasLimit, + ChainID: new(big.Int).Set(a.chainID), + BaseFee: evmOnlyBaseFee(), + BlobBaseFee: new(big.Int), + BlockHash: state.committed.blockHash, + PrevRandao: state.committed.prevRandao, } } + return executor.Call(ctx, blockCtx, msg) } -// finalizeRequest is the block identity FinalizeBlock and PrepareBlock derive from a request. -type finalizeRequest struct { - height int64 - number uint64 - timestamp uint64 - blockHash common.Hash -} - -func parseFinalizeRequest(req *abci.RequestFinalizeBlock) (finalizeRequest, error) { +func (a *evmOnlyApplication) FinalizeBlock(ctx context.Context, req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) { height := req.Header.Height if height <= 0 { - return finalizeRequest{}, fmt.Errorf("EVM-only block height must be positive: %d", height) + return nil, fmt.Errorf("EVM-only block height must be positive: %d", height) } number, ok := utils.SafeCast[uint64](height) if !ok { - return finalizeRequest{}, fmt.Errorf("EVM-only block height exceeds uint64: %d", height) + return nil, fmt.Errorf("EVM-only block height exceeds uint64: %d", height) } timestamp, ok := utils.SafeCast[uint64](req.Header.Time.Unix()) if !ok { - return finalizeRequest{}, fmt.Errorf("EVM-only block timestamp is negative: %s", req.Header.Time) - } - return finalizeRequest{ - height: height, - number: number, - timestamp: timestamp, - blockHash: common.BytesToHash(req.Hash), - }, nil -} - -// PrepareBlock decodes a block's transactions and recovers their senders before -// FinalizeBlock is called for it, so that work runs while the previous block -// executes. It may run concurrently with FinalizeBlock. Only the most recent -// prepared block is kept, and FinalizeBlock uses it only for the same height and -// hash, so preparing the wrong block costs nothing but the work: the senders -// CheckTx cached stay cached until a prepared block is consumed. Anything that -// would fail the block is left for FinalizeBlock to report; the only error -// returned is ctx ending. -func (a *evmOnlyApplication) PrepareBlock(ctx context.Context, req *abci.RequestFinalizeBlock) error { - executor, ok := a.settler.Load().Get() - if !ok { - return nil - } - block, err := parseFinalizeRequest(req) - if err != nil { - return nil - } - // Only Number and Time reach the decoded transactions (through the signer); the - // parent-derived fields are filled in by FinalizeBlock. - a.preparePhases.SetPhase("parse") - prepared, err := executor.PrepareBlock(ctx, evmonly.BlockRequest{ - Context: evmonly.BlockContext{ - Number: block.number, - Time: block.timestamp, - GasLimit: a.EvmGasLimit(), - ChainID: new(big.Int).Set(a.chainID), - BaseFee: evmOnlyBaseFee(), - BlobBaseFee: new(big.Int), - }, - Txs: req.Txs, - Senders: a.peekSenders(req.Txs), - }) - a.preparePhases.Reset() - if err != nil { - return ctx.Err() - } - for slot := range a.prepared.Lock() { - *slot = utils.Some(preparedBlock{ - height: block.height, - hash: block.blockHash, - txs: prepared.Txs, - }) - } - return nil -} - -// takePrepared returns the prepared transactions of the given block and removes -// them. A prepared block for another block is left in place. -func (a *evmOnlyApplication) takePrepared(height int64, hash common.Hash) ([]evmonly.PreparedTx, bool) { - for slot := range a.prepared.Lock() { - prepared, ok := slot.Get() - if !ok || prepared.height != height || prepared.hash != hash { - return nil, false - } - *slot = utils.None[preparedBlock]() - return prepared.txs, true - } - panic("unreachable") -} - -func (a *evmOnlyApplication) FinalizeBlock(ctx context.Context, req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) { - block, err := parseFinalizeRequest(req) - if err != nil { - return nil, err + return nil, fmt.Errorf("EVM-only block timestamp is negative: %s", req.Header.Time) } + blockHash := common.BytesToHash(req.Hash) for executor := range a.executor.Lock() { executor, ok := executor.Get() if !ok { return nil, fmt.Errorf("EVM-only block finalized before InitChain") } - return a.finalizeBlockLocked(ctx, executor, req, block) + return a.finalizeBlockLocked(ctx, executor, req, number, timestamp, blockHash) } panic("unreachable") } @@ -714,47 +499,40 @@ func (a *evmOnlyApplication) finalizeBlockLocked( ctx context.Context, executor *evmonly.Executor, req *abci.RequestFinalizeBlock, - block finalizeRequest, + number, timestamp uint64, + blockHash common.Hash, ) (*abci.ResponseFinalizeBlock, error) { - parent, err := a.beginBlock(block.height) + height := req.Header.Height + parent, err := a.beginBlock(height) if err != nil { return nil, err } - blockCtx := evmonly.BlockContext{ - Number: block.number, - Time: block.timestamp, - GasLimit: parent.gasLimit, - ChainID: new(big.Int).Set(a.chainID), - BaseFee: evmOnlyBaseFee(), - BlobBaseFee: new(big.Int), - ParentHash: parent.blockHash, - BlockHash: block.blockHash, - PrevRandao: parent.appHash, - } // Closes the stage in flight, so the gap until the next block is charged to neither. defer a.finalizePhases.Reset() - prepared, hit := a.takePrepared(block.height, block.blockHash) - a.preparedBlocks.Add(ctx, 1, otelmetric.WithAttributes(attribute.Bool("prepared", hit))) - var result *evmonly.BlockResult - if hit { - a.finalizePhases.SetPhase("take_senders") - a.forgetSenders(req.Txs) - a.finalizePhases.SetPhase("execute") - result, err = executor.ExecutePreparedBlock(ctx, evmonly.PreparedBlock{Context: blockCtx, Txs: prepared}) - } else { - a.finalizePhases.SetPhase("take_senders") - senders := a.takeSenders(req.Txs) - result, err = executeBlockPipelined(ctx, executor, a.finalizePhases, evmonly.BlockRequest{ - Context: blockCtx, - Txs: req.Txs, - Senders: senders, - }) - } + a.finalizePhases.SetPhase("take_senders") + senders := a.takeSenders(req.Txs) + // The executor's own timer breaks execution down further. + a.finalizePhases.SetPhase("execute") + result, err := executor.ExecuteBlock(ctx, evmonly.BlockRequest{ + Context: evmonly.BlockContext{ + Number: number, + Time: timestamp, + GasLimit: parent.gasLimit, + ChainID: new(big.Int).Set(a.chainID), + BaseFee: evmOnlyBaseFee(), + BlobBaseFee: new(big.Int), + ParentHash: parent.blockHash, + BlockHash: blockHash, + PrevRandao: parent.appHash, + }, + Txs: req.Txs, + Senders: senders, + }) if err != nil { - return nil, errors.Join(err, a.abandonPending(executor, block.height)) + return nil, errors.Join(err, a.abandonPending(height)) } defer result.Release() - pending, err := a.pendingCursor(block.height) + pending, err := a.pendingCursor(height) if err != nil { return nil, err } @@ -765,24 +543,6 @@ func (a *evmOnlyApplication) finalizeBlockLocked( }, nil } -// executeBlockPipelined executes the block and returns once its state commit -// has been started, leaving the commit to run while the next block executes. -// The executor lands the previous block's commit before starting this one and -// reads its changes through an overlay in the meantime, so committed-state -// readers settle through AwaitCommits rather than this returning. -// -// Preparation (decoding and recovering the senders CheckTx did not) and execution are -// timed as separate phases; the executor's own timer breaks execution down further. -func executeBlockPipelined(ctx context.Context, executor *evmonly.Executor, phases *seidbmetrics.PhaseTimer, req evmonly.BlockRequest) (*evmonly.BlockResult, error) { - phases.SetPhase("prepare") - prepared, err := executor.PrepareBlock(ctx, req) - if err != nil { - return nil, err - } - phases.SetPhase("execute") - return executor.ExecutePreparedBlock(ctx, prepared) -} - // beginBlock checks height is the next block to finalize and returns the // committed cursor it builds on. func (a *evmOnlyApplication) beginBlock(height int64) (evmOnlyCursor, error) { @@ -800,21 +560,19 @@ func (a *evmOnlyApplication) beginBlock(height int64) (evmOnlyCursor, error) { // abandonPending drops the cursor staged by a failed block unless the store // already holds that block's version, in which case the cursor is durable and -// stays pending for Commit. The in-flight commit is landed first so the store -// version is final; a commit that failed is reported alongside. -func (a *evmOnlyApplication) abandonPending(executor *evmonly.Executor, height int64) error { - commitErr := executor.AwaitCommits() +// stays pending for Commit. +func (a *evmOnlyApplication) abandonPending(height int64) error { latest, err := a.storage.SC().GetLatestVersion() if err != nil { - return errors.Join(commitErr, fmt.Errorf("read EVM-only state version: %w", err)) + return fmt.Errorf("read EVM-only state version: %w", err) } if latest >= height { - return commitErr + return nil } for state := range a.cursor.Lock() { state.pending = utils.None[evmOnlyCursor]() } - return commitErr + return nil } func (a *evmOnlyApplication) pendingCursor(height int64) (evmOnlyCursor, error) { @@ -828,12 +586,6 @@ func (a *evmOnlyApplication) pendingCursor(height int64) (evmOnlyCursor, error) panic("unreachable") } -// Commit acknowledges the finalized block as the one the chain builds on: the -// height it advances is what RPC serves as latest. Neither the block's state -// commit nor its queued receipt write is waited for here, since that would put -// the write back on the block loop, so the newest block's receipts can trail -// latest briefly. A commit that fails halts the node through the next -// FinalizeBlock, and a restart resumes from the store's own version. func (a *evmOnlyApplication) Commit(context.Context) (*abci.ResponseCommit, error) { for state := range a.cursor.Lock() { pending, ok := state.pending.Get() diff --git a/sei-tendermint/internal/evmonlyapp/app_test.go b/sei-tendermint/internal/evmonlyapp/app_test.go index 4243c9dbbb..ab230d6175 100644 --- a/sei-tendermint/internal/evmonlyapp/app_test.go +++ b/sei-tendermint/internal/evmonlyapp/app_test.go @@ -1,7 +1,6 @@ package evmonlyapp import ( - "context" "crypto/ecdsa" "encoding/binary" "errors" @@ -11,7 +10,6 @@ import ( "time" "github.com/ethereum/go-ethereum/common" - ethcore "github.com/ethereum/go-ethereum/core" ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" sdkmetric "go.opentelemetry.io/otel/sdk/metric" @@ -20,12 +18,9 @@ import ( "github.com/sei-protocol/sei-chain/giga/evmonly" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" "github.com/sei-protocol/sei-chain/sei-db/bootstrap" - "github.com/sei-protocol/sei-chain/sei-db/common/keys" seidbmetrics "github.com/sei-protocol/sei-chain/sei-db/common/metrics" - "github.com/sei-protocol/sei-chain/sei-db/proto" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" ) @@ -82,45 +77,30 @@ func newInitializedEVMOnlyTestApp(t *testing.T) abci.Application { func newEVMOnlyTestApp(t *testing.T, validators []abci.ValidatorUpdate) abci.Application { t.Helper() storage := openEVMOnlyTestStorage(t, t.TempDir()) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) app, err := NewEVMOnlyApplication(evmOnlyTestChainID, validators, storage, evmonly.NewFlatKVChangeSetEncoder(storage.SC())) require.NoError(t, err) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) return app } -// closeEVMOnlyTestApp closes storage the way the node does: after the -// application has landed every block commit it started. -func closeEVMOnlyTestApp(t *testing.T, app abci.Application, storage *bootstrap.GigaStorageManager) { - t.Helper() - settler, ok := app.(*evmOnlyApplication) - require.True(t, ok) - require.NoError(t, settler.AwaitCommits()) - require.NoError(t, storage.Close()) -} - func openEVMOnlyTestStorage(t *testing.T, home string) *bootstrap.GigaStorageManager { t.Helper() storageConfig, err := evmonly.NewValidatorStorageConfig(home, true) require.NoError(t, err) - // The store outlives the test body: the last block's commit is still landing - // when it ends, and closeEVMOnlyTestApp settles it from a cleanup, which - // runs after t.Context() is cancelled. - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - storage, err := bootstrap.NewGigaStorageManager(ctx, storageConfig) + storage, err := bootstrap.NewGigaStorageManager(t.Context(), storageConfig) require.NoError(t, err) return storage } // reopenEVMOnlyTestApp closes storage and constructs a fresh application over // the same home, the way a restarted process does. -func reopenEVMOnlyTestApp(t *testing.T, app abci.Application, storage *bootstrap.GigaStorageManager, home string) (abci.Application, *bootstrap.GigaStorageManager) { +func reopenEVMOnlyTestApp(t *testing.T, storage *bootstrap.GigaStorageManager, home string) (abci.Application, *bootstrap.GigaStorageManager) { t.Helper() - closeEVMOnlyTestApp(t, app, storage) + require.NoError(t, storage.Close()) reopened := openEVMOnlyTestStorage(t, home) - reopenedApp, err := NewEVMOnlyApplication(evmOnlyTestChainID, nil, reopened, evmonly.NewFlatKVChangeSetEncoder(reopened.SC())) + app, err := NewEVMOnlyApplication(evmOnlyTestChainID, nil, reopened, evmonly.NewFlatKVChangeSetEncoder(reopened.SC())) require.NoError(t, err) - return reopenedApp, reopened + return app, reopened } func evmOnlyTestBlock(height int64, txs ...[]byte) *abci.RequestFinalizeBlock { @@ -187,8 +167,8 @@ func TestEVMOnlyApplicationExecutesRawEthereumBlock(t *testing.T) { // Receipt writes are queued behind the block; closing the storage drains // them, so the reopened store is where the receipt is guaranteed to be. - app, storage = reopenEVMOnlyTestApp(t, app, storage, home) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) + _, storage = reopenEVMOnlyTestApp(t, storage, home) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) receiptCtx := sdk.NewContext(nil, tmproto.Header{Height: 1}, false).WithContext(t.Context()) receipt, err := storage.ReceiptDB().GetReceipt(receiptCtx, tx.Hash()) require.NoError(t, err) @@ -341,8 +321,8 @@ func TestEVMOnlyApplicationInitLastHeaderSeedsBlockTime(t *testing.T) { finalizeAndCommitEVMOnlyTestBlock(t, app, last) } - app, storage = reopenEVMOnlyTestApp(t, app, storage, home) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) + app, storage = reopenEVMOnlyTestApp(t, storage, home) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) resumed, ok := app.(*evmOnlyApplication) require.True(t, ok) @@ -380,8 +360,8 @@ func TestEVMOnlyApplicationResumesFromStorageAfterRestart(t *testing.T) { require.Equal(t, wantHashes[height], finalizeAndCommitEVMOnlyTestBlock(t, app, block(height+1))) } - app, storage = reopenEVMOnlyTestApp(t, app, storage, home) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) + app, storage = reopenEVMOnlyTestApp(t, storage, home) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) info := app.Info() require.Equal(t, int64(blocks), info.LastBlockHeight) @@ -393,10 +373,9 @@ func TestEVMOnlyApplicationResumesFromStorageAfterRestart(t *testing.T) { require.Equal(t, int64(blocks+1), app.LastBlockHeight()) } -// FinalizeBlock starts the block's state commit, and a shutdown settles it, so -// stopping after FinalizeBlock but before Commit leaves the finalized block -// durable. The restarted node must report it rather than execute it a second -// time. +// State is committed by FinalizeBlock, so a crash before Commit leaves the +// finalized block durable. The restarted node must report it rather than +// execute it a second time. func TestEVMOnlyApplicationResumesFromBlockFinalizedButNotCommitted(t *testing.T) { key, err := crypto.GenerateKey() require.NoError(t, err) @@ -414,8 +393,8 @@ func TestEVMOnlyApplicationResumesFromBlockFinalizedButNotCommitted(t *testing.T finalized, err := app.FinalizeBlock(t.Context(), block(2)) require.NoError(t, err) - app, storage = reopenEVMOnlyTestApp(t, app, storage, home) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) + app, storage = reopenEVMOnlyTestApp(t, storage, home) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) info := app.Info() require.Equal(t, int64(2), info.LastBlockHeight) @@ -436,8 +415,8 @@ func TestEVMOnlyApplicationRepeatsInitChainAfterSeedingOnly(t *testing.T) { _, err = app.InitChain(init) require.NoError(t, err) - app, storage = reopenEVMOnlyTestApp(t, app, storage, home) - t.Cleanup(func() { closeEVMOnlyTestApp(t, app, storage) }) + app, storage = reopenEVMOnlyTestApp(t, storage, home) + t.Cleanup(func() { require.NoError(t, storage.Close()) }) require.Equal(t, int64(0), app.Info().LastBlockHeight) _, err = app.InitChain(init) @@ -471,7 +450,7 @@ func TestEVMOnlyApplicationFeedsPrevRandaoThePriorAppHash(t *testing.T) { sender := crypto.PubkeyToAddress(key.PublicKey) contract := crypto.CreateAddress(sender, 0) - snapshot := app.(*evmOnlyApplication).openSettledView() + snapshot := app.(*evmOnlyApplication).storage.StateDB().OpenView() defer snapshot.Close() require.Equal(t, common.BytesToHash(prior), snapshot.GetStorage(evmOnlyStoreAddress(contract), common.Hash{})) } @@ -559,134 +538,6 @@ func TestEVMOnlyApplicationCommitsBlockWithStaleNonce(t *testing.T) { finalizeAndCommitEVMOnlyTestBlock(t, app, evmOnlyTestBlock(3)) } -// A block's state lands in the store behind FinalizeBlock. Readers of committed -// state see it before Commit, and the next block builds on it whether or not -// the store has caught up. -func TestEVMOnlyApplicationReadsSettleBehindFinalizeBlock(t *testing.T) { - app := newInitializedEVMOnlyTestApp(t) - key, err := crypto.GenerateKey() - require.NoError(t, err) - sender := crypto.PubkeyToAddress(key.PublicKey) - block := func(height int64) *abci.RequestFinalizeBlock { - return evmOnlyTestBlock(height, signedEVMOnlyTestTxFrom(t, key, evmOnlyTestChainID, uint64(height-1))) //nolint:gosec // G115: test heights are positive. - } - - for height := range int64(4) { - _, err := app.FinalizeBlock(t.Context(), block(height+1)) - require.NoError(t, err) - require.Equal(t, uint64(height+1), app.EvmNonce(sender)) //nolint:gosec // G115: test heights are positive. - require.Equal(t, height, app.LastBlockHeight()) - _, err = app.Commit(t.Context()) - require.NoError(t, err) - require.Equal(t, height+1, app.LastBlockHeight()) - } - - settler, ok := app.(*evmOnlyApplication) - require.True(t, ok) - require.NoError(t, settler.AwaitCommits()) - latest, err := settler.storage.SC().GetLatestVersion() - require.NoError(t, err) - require.Equal(t, int64(4), latest) -} - -// Nonce and balance reads answer for the block FinalizeBlock just ran, before its commit lands, -// for every account it touched; accounts it did not touch keep the funded default. -func TestEVMOnlyApplicationNonceAndBalanceReflectTheFinalizedBlock(t *testing.T) { - app := newInitializedEVMOnlyTestApp(t) - key, err := crypto.GenerateKey() - require.NoError(t, err) - sender := crypto.PubkeyToAddress(key.PublicKey) - recipient := common.HexToAddress("0x1000000000000000000000000000000000000001") - untouched := common.HexToAddress("0x2000000000000000000000000000000000000002") - - response, err := app.FinalizeBlock(t.Context(), evmOnlyTestBlock(1, signedEVMOnlyTestTxFrom(t, key, evmOnlyTestChainID, 0))) - require.NoError(t, err) - require.Len(t, response.TxResults, 1) - - require.Equal(t, uint64(1), app.EvmNonce(sender)) - gasPaid := new(big.Int).Mul(big.NewInt(evmOnlyMinGasPrice), big.NewInt(response.TxResults[0].GasUsed)) - wantSender := new(big.Int).Sub(new(big.Int).Sub(new(big.Int).Set(evmOnlyBaseBalance), big.NewInt(1)), gasPaid) - senderBalance := app.EvmBalance(sender, nil) - require.Equal(t, wantSender, senderBalance.ToBig()) - recipientBalance := app.EvmBalance(recipient, nil) - require.Equal(t, new(big.Int).Add(new(big.Int).Set(evmOnlyBaseBalance), big.NewInt(1)), recipientBalance.ToBig()) - require.Equal(t, uint64(0), app.EvmNonce(untouched)) - untouchedBalance := app.EvmBalance(untouched, nil) - require.Equal(t, evmOnlyBaseBalance, untouchedBalance.ToBig()) - - _, err = app.Commit(t.Context()) - require.NoError(t, err) - require.Equal(t, uint64(1), app.EvmNonce(sender)) -} - -// Once a commit has failed, nonce and balance reads fall back to the store, which stays at the -// last version that landed. -func TestEVMOnlyApplicationNonceAndBalanceFallBackToTheStoreAfterAFailedCommit(t *testing.T) { - storage := openEVMOnlyTestStorage(t, t.TempDir()) - app, err := NewEVMOnlyApplication(evmOnlyTestChainID, nil, storage, unwritableEVMChangeSetEncoder) - require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, storage.Close()) }) - _, err = app.InitChain(evmOnlyTestInitChain()) - require.NoError(t, err) - settler, ok := app.(*evmOnlyApplication) - require.True(t, ok) - - raw, sender := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - finalizeAndCommitEVMOnlyTestBlock(t, app, evmOnlyTestBlock(1, raw)) - require.Error(t, settler.AwaitCommits()) - - require.Equal(t, uint64(0), app.EvmNonce(sender)) - balance := app.EvmBalance(sender, nil) - require.Equal(t, evmOnlyBaseBalance, balance.ToBig()) -} - -// unwritableEVMChangeSetEncoder encodes every block with an EVM pair the store -// refuses to apply, so the block executes and encodes cleanly and its commit is -// the first thing that fails. -func unwritableEVMChangeSetEncoder(evmonly.StateChangeSet) ([]*proto.NamedChangeSet, error) { - return []*proto.NamedChangeSet{{ - Name: keys.EVMStoreKey, - Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{{Key: nil, Value: []byte{1}}}}, - }}, nil -} - -// A block's commit lands behind FinalizeBlock: the block whose commit fails is -// still finalized and committed, and the failure surfaces from the next -// FinalizeBlock, from read-only calls, and from settling the store, while the -// store itself stays at the last version that landed. -func TestEVMOnlyApplicationSurfacesAFailedCommitFromTheNextBlock(t *testing.T) { - storage := openEVMOnlyTestStorage(t, t.TempDir()) - app, err := NewEVMOnlyApplication(evmOnlyTestChainID, nil, storage, unwritableEVMChangeSetEncoder) - require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, storage.Close()) }) - _, err = app.InitChain(evmOnlyTestInitChain()) - require.NoError(t, err) - settler, ok := app.(*evmOnlyApplication) - require.True(t, ok) - - // Block 1 is unwritable, yet it finalizes and commits: the write has not - // been waited for. The synchronous path would fail here. - raw, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - finalizeAndCommitEVMOnlyTestBlock(t, app, evmOnlyTestBlock(1, raw)) - require.Equal(t, int64(1), app.LastBlockHeight()) - - // The failed write is reported by the next block, and stays reported. - _, err = app.FinalizeBlock(t.Context(), evmOnlyTestBlock(2)) - require.Error(t, err) - require.Error(t, settler.AwaitCommits()) - _, err = settler.EvmCall(t.Context(), ðcore.Message{GasLimit: 21_000, GasPrice: new(big.Int), Value: new(big.Int)}) - require.Error(t, err) - - // The block that failed to finalize left nothing staged, and the store never - // moved past genesis. - _, err = app.Commit(t.Context()) - require.Error(t, err) - require.Equal(t, int64(1), app.LastBlockHeight()) - latest, err := storage.SC().GetLatestVersion() - require.NoError(t, err) - require.Equal(t, int64(0), latest) -} - // TestHashRawTxsMatchesKeccak256Hash pins hashRawTxs to crypto.Keccak256Hash, which keys the sender cache. func TestHashRawTxsMatchesKeccak256Hash(t *testing.T) { for _, count := range []int{0, 1, 2, 17, 64, 65, 200, 1848} { @@ -734,165 +585,8 @@ func TestEVMOnlyApplicationTimesEveryFinalizeBlockPhase(t *testing.T) { } } } - for _, want := range []string{"take_senders", "prepare", "execute", "tx_results"} { + for _, want := range []string{"take_senders", "execute", "tx_results"} { _, ok := phases[want] require.True(t, ok, "phase %q not recorded", want) } } - -// preparedBlockCounts reads evmonly_finalize_prepared_blocks_total by its prepared label. -func preparedBlockCounts(t *testing.T, reader *sdkmetric.ManualReader) map[bool]int64 { - t.Helper() - var rm metricdata.ResourceMetrics - require.NoError(t, reader.Collect(t.Context(), &rm)) - counts := map[bool]int64{} - for _, scope := range rm.ScopeMetrics { - for _, m := range scope.Metrics { - if m.Name != "evmonly_finalize_prepared_blocks_total" { - continue - } - sum, ok := m.Data.(metricdata.Sum[int64]) - require.True(t, ok) - for _, point := range sum.DataPoints { - prepared, ok := point.Attributes.Value("prepared") - require.True(t, ok) - counts[prepared.AsBool()] = point.Value - } - } - } - return counts -} - -// A block decoded ahead of FinalizeBlock has to execute to the same result as one -// decoded inside it, and a block nobody prepared still has to execute. -func TestEVMOnlyApplicationExecutesAPreparedBlockLikeAnUnpreparedOne(t *testing.T) { - prepared := newInitializedEVMOnlyTestApp(t) - preparedApp, ok := prepared.(*evmOnlyApplication) - require.True(t, ok) - reader := sdkmetric.NewManualReader() - preparedApp.preparedBlocks = newPreparedBlocksCounter( - sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter(finalizeMeterName), - ) - unprepared := newInitializedEVMOnlyTestApp(t) - - first, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - second, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - for height, txs := range [][][]byte{{first}, {second}} { - block := evmOnlyTestBlock(int64(height)+1, txs...) - if height == 0 { - require.NoError(t, preparedApp.PrepareBlock(t.Context(), block)) - } - preparedResp, err := prepared.FinalizeBlock(t.Context(), block) - require.NoError(t, err) - unpreparedResp, err := unprepared.FinalizeBlock(t.Context(), block) - require.NoError(t, err) - require.Equal(t, unpreparedResp.AppHash, preparedResp.AppHash) - require.Equal(t, len(unpreparedResp.TxResults), len(preparedResp.TxResults)) - for i := range preparedResp.TxResults { - require.Equal(t, unpreparedResp.TxResults[i].Code, preparedResp.TxResults[i].Code) - require.Equal(t, unpreparedResp.TxResults[i].GasUsed, preparedResp.TxResults[i].GasUsed) - } - _, err = prepared.Commit(t.Context()) - require.NoError(t, err) - _, err = unprepared.Commit(t.Context()) - require.NoError(t, err) - } - counts := preparedBlockCounts(t, reader) - require.Equal(t, int64(1), counts[true]) - require.Equal(t, int64(1), counts[false]) -} - -// A prepared block is only used for the block it was prepared for: one with the -// same height but another hash is decoded again, and the prepared one is kept -// for its own block. -func TestEVMOnlyApplicationIgnoresAPreparedBlockForAnotherHash(t *testing.T) { - prepared := newInitializedEVMOnlyTestApp(t) - preparedApp, ok := prepared.(*evmOnlyApplication) - require.True(t, ok) - unprepared := newInitializedEVMOnlyTestApp(t) - - other, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - otherBlock := evmOnlyTestBlock(1, other) - otherBlock.Hash = crypto.Keccak256([]byte("other-block")) - require.NoError(t, preparedApp.PrepareBlock(t.Context(), otherBlock)) - - raw, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - block := evmOnlyTestBlock(1, raw) - preparedHash := finalizeAndCommitEVMOnlyTestBlock(t, prepared, block) - unpreparedHash := finalizeAndCommitEVMOnlyTestBlock(t, unprepared, block) - require.Equal(t, unpreparedHash, preparedHash) - _, ok = preparedApp.takePrepared(1, common.BytesToHash(otherBlock.Hash)) - require.True(t, ok) -} - -// A block PrepareBlock cannot decode is reported by FinalizeBlock, the same as -// when nobody prepared it. -func TestEVMOnlyApplicationReportsAnUndecodableBlockFromFinalizeBlock(t *testing.T) { - prepared := newInitializedEVMOnlyTestApp(t) - preparedApp, ok := prepared.(*evmOnlyApplication) - require.True(t, ok) - unprepared := newInitializedEVMOnlyTestApp(t) - - block := evmOnlyTestBlock(1, []byte("not a transaction")) - require.NoError(t, preparedApp.PrepareBlock(t.Context(), block)) - _, preparedErr := prepared.FinalizeBlock(t.Context(), block) - require.Error(t, preparedErr) - _, unpreparedErr := unprepared.FinalizeBlock(t.Context(), block) - require.Error(t, unpreparedErr) -} - -// Preparing before InitChain is a no-op rather than a failure. -func TestEVMOnlyApplicationPrepareBlockBeforeInitChainIsANoOp(t *testing.T) { - app := newEVMOnlyTestApp(t, nil) - evmOnlyApp, ok := app.(*evmOnlyApplication) - require.True(t, ok) - raw, _ := signedEVMOnlyTestTx(t, evmOnlyTestChainID, 0) - require.NoError(t, evmOnlyApp.PrepareBlock(t.Context(), evmOnlyTestBlock(1, raw))) -} - -// PrepareBlock for the next block runs while FinalizeBlock runs the current one, -// and both blocks come out as if they had been finalized alone. -func TestEVMOnlyApplicationPreparesTheNextBlockWhileFinalizingTheCurrentOne(t *testing.T) { - prepared := newInitializedEVMOnlyTestApp(t) - preparedApp, ok := prepared.(*evmOnlyApplication) - require.True(t, ok) - unprepared := newInitializedEVMOnlyTestApp(t) - - key, err := crypto.GenerateKey() - require.NoError(t, err) - var blocks []*abci.RequestFinalizeBlock - for height := range int64(4) { - var txs [][]byte - for i := range 8 { - txs = append(txs, signedEVMOnlyTestTxFrom(t, key, evmOnlyTestChainID, uint64(height)*8+uint64(i))) //nolint:gosec // G115: small test counters. - } - blocks = append(blocks, evmOnlyTestBlock(height+1, txs...)) - } - - hashes := make([][]byte, len(blocks)) - require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { - for i, block := range blocks { - var next *abci.RequestFinalizeBlock - if i+1 < len(blocks) { - next = blocks[i+1] - } - resp, err := scope.Run1(ctx, func(ctx context.Context, s scope.Scope) (*abci.ResponseFinalizeBlock, error) { - if next != nil { - s.Spawn(func() error { return preparedApp.PrepareBlock(ctx, next) }) - } - return prepared.FinalizeBlock(ctx, block) - }) - if err != nil { - return err - } - if _, err := prepared.Commit(ctx); err != nil { - return err - } - hashes[i] = resp.AppHash - } - return nil - })) - for i, block := range blocks { - require.Equal(t, finalizeAndCommitEVMOnlyTestBlock(t, unprepared, block), hashes[i]) - } -} diff --git a/sei-tendermint/internal/mempool/tx.go b/sei-tendermint/internal/mempool/tx.go index 2a325995ab..7b53d436ba 100644 --- a/sei-tendermint/internal/mempool/tx.go +++ b/sei-tendermint/internal/mempool/tx.go @@ -126,9 +126,6 @@ type txStoreInner struct { byEvmHash map[common.Hash]*WrappedTx byNonce map[evmAddrNonce]*WrappedTx accounts map[common.Address]*evmAccount - // Incremented whenever accounts is reset, so account state fetched outside the lock is only - // installed against the height it was fetched for. - accountsEpoch uint64 softLimit txCounter hardLimit txCounter @@ -206,7 +203,6 @@ func (s *txStore) Clear() { inner.byEvmHash = map[common.Hash]*WrappedTx{} inner.byNonce = map[evmAddrNonce]*WrappedTx{} inner.accounts = map[common.Address]*evmAccount{} - inner.accountsEpoch++ inner.state.Store(txStoreState{}) s.readyTxs.Clear() } @@ -366,8 +362,7 @@ func (s *txStore) insert(inner *txStoreInner, wtx *WrappedTx, recordAdded bool) // Fetch the evm account state. account, ok := inner.accounts[evm.address] if !ok { - // Insert prefetches the account outside the mutex; this only runs when that prefetch - // was discarded, or for txs reinserted by compact. + // TODO(gprusak): consider whether we should move these queries out of the mutex. b := s.app.EvmBalance(evm.address, evm.seiAddress) n := s.app.EvmNonce(evm.address) account = &evmAccount{b, n, n} @@ -511,53 +506,10 @@ func (inner *txStoreInner) inInclusionOrder() []*WrappedTx { return res } -// prefetchedAccount is evm account state fetched from the app outside the store lock, tagged with -// the accounts epoch it is valid for. -type prefetchedAccount struct { - address common.Address - epoch uint64 - account *evmAccount -} - -// prefetchAccount fetches the evm account state of wtx's sender without holding the store lock, when -// the store does not know the account yet. Reading the app can wait on the state store, and doing -// so under the lock would stall every other mempool operation. -func (s *txStore) prefetchAccount(wtx *WrappedTx) utils.Option[prefetchedAccount] { - evm, ok := wtx.evm.Get() - if !ok { - return utils.None[prefetchedAccount]() - } - var epoch uint64 - for inner := range s.inner.RLock() { - if _, ok := inner.accounts[evm.address]; ok { - return utils.None[prefetchedAccount]() - } - epoch = inner.accountsEpoch - } - b := s.app.EvmBalance(evm.address, evm.seiAddress) - n := s.app.EvmNonce(evm.address) - return utils.Some(prefetchedAccount{address: evm.address, epoch: epoch, account: &evmAccount{b, n, n}}) -} - -// installPrefetched adds a prefetched account to inner unless the accounts were reset since it was -// fetched or another inserter got there first. -func installPrefetched(inner *txStoreInner, prefetched utils.Option[prefetchedAccount]) { - p, ok := prefetched.Get() - if !ok || p.epoch != inner.accountsEpoch { - return - } - if _, ok := inner.accounts[p.address]; ok { - return - } - inner.accounts[p.address] = p.account -} - // Inserts a new transaction to txStore. // txStore takes ownership of wtx. func (s *txStore) Insert(wtx *WrappedTx) error { - prefetched := s.prefetchAccount(wtx) for inner := range s.inner.Lock() { - installPrefetched(inner, prefetched) if err := s.insert(inner, wtx, true); err != nil { return err } @@ -586,7 +538,6 @@ func (s *txStore) compact(inner *txStoreInner, clearAccounts bool) { inner.byNonce = map[evmAddrNonce]*WrappedTx{} if clearAccounts { inner.accounts = map[common.Address]*evmAccount{} - inner.accountsEpoch++ } for _, account := range inner.accounts { account.nextNonce = account.firstNonce diff --git a/sei-tendermint/internal/mempool/tx_test.go b/sei-tendermint/internal/mempool/tx_test.go index 8b8a95b770..79d7c87bd6 100644 --- a/sei-tendermint/internal/mempool/tx_test.go +++ b/sei-tendermint/internal/mempool/tx_test.go @@ -1,7 +1,6 @@ package mempool import ( - "context" "errors" "fmt" "testing" @@ -13,7 +12,6 @@ import ( "github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" "github.com/sei-protocol/sei-chain/sei-tendermint/types" ) @@ -791,111 +789,3 @@ func TestTxStore_InsertCompactionKeepsReadyListInSync(t *testing.T) { require.ElementsMatch(t, toTxs(expected), listed) } } - -// gatedNonceApp is an evmNonceApp whose nonce reads block until released, so a test can hold an -// account fetch in flight and observe what the store does meanwhile. -type gatedNonceApp struct { - *evmNonceApp - inFlight utils.AtomicSend[int] - release utils.AtomicRecv[bool] -} - -// EvmNonce reads the nonce first and only then blocks, so the value it returns is the one that -// was current when the fetch began. -func (a *gatedNonceApp) EvmNonce(addr common.Address) uint64 { - nonce := a.evmNonceApp.EvmNonce(addr) - a.inFlight.Store(a.inFlight.Load() + 1) - if _, err := a.release.Wait(context.Background(), func(v bool) bool { return v }); err != nil { - panic(err) - } - return nonce -} - -func evmTxForTest(rng utils.Rng, address common.Address, nonce uint64) *WrappedTx { - return &WrappedTx{ - hashedTx: newHashedTx(utils.GenBytes(rng, 32)), - timestamp: time.Now(), - priority: 1, - gasWanted: 1, - estimatedGas: 1, - evm: utils.Some(evmTx{ - address: address, - seiAddress: address.Bytes(), - hash: genEvmHash(rng), - nonce: nonce, - }), - } -} - -// Fetching a first-seen account from the app must not hold the store lock: other inserts and -// reads proceed while the fetch is in flight. -func TestTxStore_InsertFetchesFirstSeenAccountOutsideTheLock(t *testing.T) { - rng := utils.TestRng() - release := utils.NewAtomicSend(false) - app := &gatedNonceApp{evmNonceApp: newEVMNonceApp(), inFlight: utils.NewAtomicSend(0), release: release.Subscribe()} - txStore := NewTxStore(TestConfig(), proxy.New(app)) - cold := common.BytesToAddress(utils.GenBytes(rng, 20)) - warm := common.BytesToAddress(utils.GenBytes(rng, 20)) - - // warm is known to the store before any fetch is gated; its fetch does not count. - release.Store(true) - require.NoError(t, txStore.Insert(evmTxForTest(rng, warm, 0))) - release.Store(false) - app.inFlight.Store(0) - - coldTx := evmTxForTest(rng, cold, 0) - warmTx := evmTxForTest(rng, warm, 1) - require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { - s.Spawn(func() error { return txStore.Insert(coldTx) }) - inFlight := app.inFlight.Subscribe() - if _, err := inFlight.Wait(ctx, func(n int) bool { return n == 1 }); err != nil { - return err - } - // Both a read and a write of the store complete while the cold fetch is blocked. - if got := txStore.NextNonce(warm); got != 1 { - return fmt.Errorf("NextNonce(warm) = %d, want 1", got) - } - if err := txStore.Insert(warmTx); err != nil { - return err - } - release.Store(true) - return nil - })) - _, ok := txStore.ByHash(coldTx.Hash()) - require.True(t, ok) - _, ok = txStore.ByHash(warmTx.Hash()) - require.True(t, ok) -} - -// Account state fetched before an Update belongs to the previous height and must not be installed -// after it: the insert is judged against the account nonce of the new height. -func TestTxStore_InsertDiscardsAccountFetchedBeforeUpdate(t *testing.T) { - rng := utils.TestRng() - release := utils.NewAtomicSend(false) - app := &gatedNonceApp{evmNonceApp: newEVMNonceApp(), inFlight: utils.NewAtomicSend(0), release: release.Subscribe()} - txStore := NewTxStore(TestConfig(), proxy.New(app)) - sender := common.BytesToAddress(utils.GenBytes(rng, 20)) - - staleTx := evmTxForTest(rng, sender, 0) - err := scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error { - s.Spawn(func() error { return txStore.Insert(staleTx) }) - inFlight := app.inFlight.Subscribe() - if _, err := inFlight.Wait(ctx, func(n int) bool { return n == 1 }); err != nil { - return err - } - // Nonce 0 is mined while the fetch that read it as pending is still in flight. - app.markMined(sender) - txStore.Update(updateSpec{ - Now: time.Now(), - Height: 1, - TxResults: map[types.TxHash]bool{}, - Constraints: NopTxConstraints(), - NewPriorities: map[types.TxHash]int64{}, - }) - release.Store(true) - return nil - }) - require.ErrorIs(t, err, errOldNonce) - _, ok := txStore.ByHash(staleTx.Hash()) - require.False(t, ok) -} diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index b263c5c63e..f71ba4cf31 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -224,11 +224,22 @@ func (r *gigaRouterCommon) translateGlobalBlock(gb *atypes.GlobalBlock) *coretyp } } -// finalizeRequest builds the FinalizeBlock request for a global block, except -// for the header's ProposerAddress, which depends on the committed state. -func (r *gigaRouterCommon) finalizeRequest(b *atypes.GlobalBlock) *abci.RequestFinalizeBlock { +func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlock, hashVault hashvault.HashVault) (*abci.ResponseCommit, error) { + app := r.app hash := b.Header.Hash() - return &abci.RequestFinalizeBlock{ + var proposerAddress types.Address + if vals := app.GetValidators(); len(vals) > 0 { + // Deterministically select a proposer from the app's validator committee. + // We need it so that app does not emit error logs. + proposer := slices.MinFunc(vals, func(a, b abci.ValidatorUpdate) int { return a.PubKey.Compare(b.PubKey) }) + key, err := crypto.PubKeyFromProto(proposer.PubKey) + if err != nil { + return nil, fmt.Errorf("crypto.PubKeyFromProto(): %w", err) + } + proposerAddress = key.Address() + } + + resp, err := app.FinalizeBlock(ctx, &abci.RequestFinalizeBlock{ Txs: b.Payload.Txs(), // Empty DecidedLastCommit does not indicate missing votes. DecidedLastCommit: abci.CommitInfo{}, @@ -240,45 +251,11 @@ func (r *gigaRouterCommon) finalizeRequest(b *atypes.GlobalBlock) *abci.RequestF ChainID: r.cfg.GenDoc.ChainID, Height: int64(b.GlobalNumber), // nolint:gosec // different representations of the same value Time: b.Timestamp, + // WARNING: the reward distribution has corner cases where it forgets the proposer, + // because reward is distributed with a delay. This is not our problem here though. + ProposerAddress: proposerAddress, }).ToProto(), - } -} - -// proposerAddress returns the proposer of the next block, selected from the -// app's current validator committee. -func (r *gigaRouterCommon) proposerAddress() (types.Address, error) { - vals := r.app.GetValidators() - if len(vals) == 0 { - return nil, nil - } - // Deterministically select a proposer from the app's validator committee. - // We need it so that app does not emit error logs. - proposer := slices.MinFunc(vals, func(a, b abci.ValidatorUpdate) int { return a.PubKey.Compare(b.PubKey) }) - key, err := crypto.PubKeyFromProto(proposer.PubKey) - if err != nil { - return nil, fmt.Errorf("crypto.PubKeyFromProto(): %w", err) - } - return key.Address(), nil -} - -// fetchedBlock is a global block with the FinalizeBlock request built for it. -type fetchedBlock struct { - block *atypes.GlobalBlock - req *abci.RequestFinalizeBlock -} - -func (r *gigaRouterCommon) executeBlock(ctx context.Context, f fetchedBlock, hashVault hashvault.HashVault) (*abci.ResponseCommit, error) { - app := r.app - b := f.block - // Read from the app's state on the execute loop, after the previous block's Commit. - // WARNING: the reward distribution has corner cases where it forgets the proposer, - // because reward is distributed with a delay. This is not our problem here though. - proposer, err := r.proposerAddress() - if err != nil { - return nil, err - } - f.req.Header.ProposerAddress = proposer - resp, err := app.FinalizeBlock(ctx, f.req) + }) if err != nil { return nil, fmt.Errorf("app.FinalizeBlock(): %w", err) } @@ -498,61 +475,39 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { } } - return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { - // Blocks are fetched and prepared one ahead of execution: the unbuffered - // channel lets the fetcher hold block n+1, already prepared, while the - // loop below executes block n. - blocks := make(chan fetchedBlock) - s.Spawn(func() error { - for n := next; ; n += 1 { - b, err := r.data.GlobalBlock(ctx, n) - if err != nil { - return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) - } - req := r.finalizeRequest(b) - if err := app.PrepareBlock(ctx, req); err != nil { - return fmt.Errorf("app.PrepareBlock(%v): %w", n, err) - } - if err := utils.Send(ctx, blocks, fetchedBlock{block: b, req: req}); err != nil { - return err - } - } - }) - for { - gigametrics.SetPhase(gigametrics.PhaseConsensus) - f, err := utils.Recv(ctx, blocks) - if err != nil { - return err - } - n := f.block.GlobalNumber - gigametrics.SetPhase(gigametrics.PhaseExecution) - commitResp, err := r.executeBlock(ctx, f, hashVault) - if err != nil { - return fmt.Errorf("r.executeBlock(%v): %w", n, err) - } - pruneBefore, ok := utils.SafeCast[atypes.GlobalBlockNumber](commitResp.RetainHeight) - if !ok { - return fmt.Errorf("invalid commitResp.RetainHeight = %v", commitResp.RetainHeight) - } - gigametrics.SetStoragePhase(gigametrics.StoragePhasePruneData) - if err := r.data.PruneBefore(pruneBefore); err != nil { - return fmt.Errorf("r.data.PruneBefore(%v): %w", pruneBefore, err) - } - // Align the vault's retention with the data layer's prune boundary. - gigametrics.SetStoragePhase(gigametrics.StoragePhasePruneVault) - if err := hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { - // A canceled context just means we're shutting down between a successful executeBlock - // and this prune; that's benign, not a prune failure, so don't alarm operators. - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { - logger.Info("hashvault prune aborted by context cancellation during shutdown", - "prune_before", pruneBefore, "err", err) - } else { - logger.Error("failed to prune hashvault", "prune_before", pruneBefore, "err", err) - } + for n := next; ; n += 1 { + gigametrics.SetPhase(gigametrics.PhaseConsensus) + b, err := r.data.GlobalBlock(ctx, n) + if err != nil { + return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) + } + gigametrics.SetPhase(gigametrics.PhaseExecution) + commitResp, err := r.executeBlock(ctx, b, hashVault) + if err != nil { + return fmt.Errorf("r.executeBlock(%v): %w", n, err) + } + pruneBefore, ok := utils.SafeCast[atypes.GlobalBlockNumber](commitResp.RetainHeight) + if !ok { + return fmt.Errorf("invalid commitResp.RetainHeight = %v", commitResp.RetainHeight) + } + gigametrics.SetStoragePhase(gigametrics.StoragePhasePruneData) + if err := r.data.PruneBefore(pruneBefore); err != nil { + return fmt.Errorf("r.data.PruneBefore(%v): %w", pruneBefore, err) + } + // Align the vault's retention with the data layer's prune boundary. + gigametrics.SetStoragePhase(gigametrics.StoragePhasePruneVault) + if err := hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { + // A canceled context just means we're shutting down between a successful executeBlock + // and this prune; that's benign, not a prune failure, so don't alarm operators. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + logger.Info("hashvault prune aborted by context cancellation during shutdown", + "prune_before", pruneBefore, "err", err) + } else { + logger.Error("failed to prune hashvault", "prune_before", pruneBefore, "err", err) } - gigametrics.EndStoragePhase() } - }) + gigametrics.EndStoragePhase() + } } // dialAndRunConn dials a peer, handshakes as a SeiGiga connection, diff --git a/sei-tendermint/internal/p2p/giga_router_testhelper_test.go b/sei-tendermint/internal/p2p/giga_router_testhelper_test.go index 731f8d6093..b5bc6e3533 100644 --- a/sei-tendermint/internal/p2p/giga_router_testhelper_test.go +++ b/sei-tendermint/internal/p2p/giga_router_testhelper_test.go @@ -1,7 +1,6 @@ package p2p import ( - "bytes" "context" "crypto/sha256" "encoding/json" @@ -153,39 +152,6 @@ func (a *testApp) FinalizeBlock(_ context.Context, req *abci.RequestFinalizeBloc panic("unreachable") } -// CheckBlocks verifies the FinalizeBlock requests the router delivered: heights -// are contiguous from InitialHeight and every header names a current validator -// as proposer. -func (s *testAppState) CheckBlocks() error { - init, ok := s.Init.Get() - if !ok { - return fmt.Errorf("app not initialized") - } - for i, b := range s.Blocks { - if want := init.InitialHeight + int64(i); b.Header.Height != want { - return fmt.Errorf("blocks[%v].Height = %v, want %v", i, b.Header.Height, want) - } - if err := checkProposer(b.Header.ProposerAddress, s.Validators); err != nil { - return fmt.Errorf("blocks[%v]: %w", i, err) - } - } - return nil -} - -// checkProposer verifies that the proposer is one of the given validators. -func checkProposer(proposer types.Address, vals []abci.ValidatorUpdate) error { - for _, val := range vals { - key, err := crypto.PubKeyFromProto(val.PubKey) - if err != nil { - return fmt.Errorf("crypto.PubKeyFromProto(): %w", err) - } - if bytes.Equal(key.Address(), proposer) { - return nil - } - } - return fmt.Errorf("proposer %X is not a current validator", proposer) -} - func (a *testApp) Commit(context.Context) (*abci.ResponseCommit, error) { for state, ctrl := range a.state.Lock() { if state.Committed { diff --git a/sei-tendermint/internal/p2p/giga_router_validator_test.go b/sei-tendermint/internal/p2p/giga_router_validator_test.go index cf3c71cdf4..aa216a0c84 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator_test.go +++ b/sei-tendermint/internal/p2p/giga_router_validator_test.go @@ -149,7 +149,6 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { } // Nodes should agree on the final state. want := apps[0].Snapshot() - require.NoError(t, want.CheckBlocks(), "CheckBlocks") for i, app := range apps { t.Logf("app[%v]", i) require.NoError(t, utils.TestDiff(want, app.Snapshot()), "state mismatch app[%v]", i) diff --git a/sei-tendermint/internal/proxy/proxy.go b/sei-tendermint/internal/proxy/proxy.go index c8ba5b68a1..998733e63f 100644 --- a/sei-tendermint/internal/proxy/proxy.go +++ b/sei-tendermint/internal/proxy/proxy.go @@ -62,41 +62,6 @@ func (app *Proxy) EvmChainID() uint64 { return app.app.EvmChainID() } -// commitSettler is implemented by applications whose block state commits land -// behind FinalizeBlock. -type commitSettler interface { - AwaitCommits() error -} - -// AwaitCommits blocks until every block the wrapped application has finalized -// is in its store, and reports the first commit that failed. It returns nil for -// an application that commits synchronously. -func (app *Proxy) AwaitCommits() error { - defer addTimeSample(Global.MethodTimingAt("await_commits", "sync"))() - settler, ok := app.app.(commitSettler) - if !ok { - return nil - } - return settler.AwaitCommits() -} - -// blockPreparer is implemented by applications that can do the stateless part of -// FinalizeBlock for a block before it is called. -type blockPreparer interface { - PrepareBlock(context.Context, *types.RequestFinalizeBlock) error -} - -// PrepareBlock lets the wrapped application decode a block ahead of its -// FinalizeBlock. It is a no-op for an application that does not prepare blocks. -func (app *Proxy) PrepareBlock(ctx context.Context, req *types.RequestFinalizeBlock) error { - defer addTimeSample(Global.MethodTimingAt("prepare_block", "sync"))() - preparer, ok := app.app.(blockPreparer) - if !ok { - return nil - } - return preparer.PrepareBlock(ctx, req) -} - // evmCaller is implemented by applications that can run a read-only EVM call // against their current state. type evmCaller interface { diff --git a/sei-tendermint/node/node.go b/sei-tendermint/node/node.go index f535f2faaa..5dbfd64f20 100644 --- a/sei-tendermint/node/node.go +++ b/sei-tendermint/node/node.go @@ -769,15 +769,11 @@ func (n *nodeImpl) OnStop() { } // closeGigaStorage closes the manager-owned storage or standalone Autobahn -// block store at most once. The application's in-flight block commits are -// landed first, so the stores are not closed under a write. +// block store at most once. func (n *nodeImpl) closeGigaStorage() error { var err error n.gigaStorageCloseOnce.Do(func() { if manager, ok := n.gigaStorageManager.Get(); ok { - if settleErr := n.rpcEnv.App.AwaitCommits(); settleErr != nil { - logger.Error("failed to settle block commits before closing Giga storage", "err", settleErr) - } if err = manager.Close(); err != nil { logger.Error("failed to close Giga storage manager", "err", err) }