erigontech / erigontech/erigon
execution: extract apply loop into event stream with fan-out consumers
- Dominant language
- Go
- Stars
- 3.6k
- Forks
- 1.5k
- Avg merge
- 1d 16h
- Merged PRs (30d)
- 455
Description
## Background
The parallel execution apply loop (`exec3_parallel.go`) is a single goroutine consuming `applyResults` and performing all post-execution work serially: state writes, index/receipt writes, accumulator notifications, commitment, and block validation. These are independent concerns that should be separated — both for clarity and to enable independent load management and extensibility.
Currently:
```
Executor goroutines → applyResults channel → single apply goroutine:
├── applyVersionedWrites (state → SharedDomains → MDBX) ← critical path
├── applyLogsAndTraces4 (traces, log indices, receipts) ← PBC servicing
├── ComputeCommitment (per-step) ← trie hashing
├── accumulator.ChangeAccount/Code/Storage (txpool diffs) ← notifications
├── notifications.RecentReceipts.Add ← notifications
└── block post-validation ← validation
```
The accumulator is threaded *through* the state writer (`versionedWriteCollector`) during execution, but it's consuming the same `VersionedWrites` data that already flows through `applyResults`. Similarly, `applyLogsAndTraces4` (traces, log indices, receipts) is PBC servicing that doesn't belong on the state-write critical path.
## Goal
Extract the apply loop into a dispatcher that fans out to independent consumers of a unified event stream.
```
applyResults → dispatcher
├── stateConsumer (applyVersionedWrites → SharedDomains) ← critical path
├── indexConsumer (traces, logs, receipts → MDBX indices) ← can lag behind
├── notifyConsumer (accumulator → txpool notifications) ← can lag behind
└── commitConsumer (async commitment → trie hash) ← separate goroutine
```
This gives us:
- **Load management**: index and notification work no longer blocks the state-write critical path
- **Extensibility**: new consumers (metrics, audit log, external subscribers) can tap the stream without modifying the apply loop
- **Simplification**: each consumer has a single responsibility
- **BAL-driven path enablement**: switching between "write" and "compare" modes (#19791) is purely about the state consumer, with no accumulator side effects
## Incremental steps
### Step 1 — Split `ApplyTxState` into state + index concerns
Split the current `ApplyTxState` method into two methods:
- `ApplyStateWrites(ctx, roTx, blockNum, txNum, writes, balanceIncreases, rules)` — just `applyVersionedWrites`
- `ApplyTxIndexes(roTx, txNum, receipt, blobGas, logs, traceFroms, traceTos, historyExecution)` — just `applyLogsAndTraces4`
Call both from the apply loop. No goroutine changes yet — just clean separation in the code.
**Files**: `execution/state/rw_v3.go`, `execution/stagedsync/exec3_parallel.go`
### Step 2 — Decouple accumulator from the state writer
Remove the `accumulator *shards.Accumulator` field from `versionedWriteCollector`. Instead, drive the accumulator from the apply loop by extracting account/code/storage changes directly from `txResult.writes` (VersionedWrites).
The data is already there:
- `AddressPath` writes → `ChangeAccount`
- `CodePath` writes → `ChangeCode`
- `StoragePath` writes → `ChangeStorage`
`StartChange` is already called from the apply loop (line 656), not from the writer. This change means:
- `NewVersionedWriteCollector(rs)` — no accumulator parameter
- The accumulator becomes a consumer of the apply stream, not a participant in the write path
- Execution code no longer threads notification concerns
**Files**: `execution/state/rw_v3.go`, `execution/stagedsync/exec3_parallel.go`, `execution/stagedsync/exec3_2cache_test.go`, `execution/protocol/rules/aura/aura_test.go`
### Step 3 — Fan-out dispatcher
Restructure the apply loop into a dispatcher pattern:
```go
type applyEvent struct {
blockNum uint64
txNum uint64
writes VersionedWrites
receipt *types.Receipt
logs []*types.Log
traceFroms map[accounts.Address]struct{}
traceTos map[accounts.Address]struct{}
// ... other fields
}
```
The dispatcher reads from `applyResults`, constructs `applyEvent`s, and fans out to registered consumers. State writes remain synchronous (critical path). Index writes, accumulator notifications, and commitment can be asynchronous consumers with their own back-pressure.
**Files**: `execution/stagedsync/exec3_parallel.go` (new dispatcher), potentially new `execution/stagedsync/apply_consumers.go`
## Ordering and dependencies
- **Step 1**: standalone, can be done now
- **Step 2**: depends on Step 1 (cleaner with the split). Also reverses the accumulator wiring added in PR #19711 — should wait for #19711 to merge first
- **Step 3**: depends on Step 2. Also aligns with #19791 (commitment pipelining) — the commitConsumer IS the async commitment from #19791
## Relationship to other work
- **#19711** (ibs-2cache phase 2+3): added accumulator to `versionedWriteCollector`. Step 2 here reverses that by moving the accumulator out of the writer entirely.
- **#19701** (phase 4 — direct `VersionedWrites → SharedDomains`): simplifies the state consumer further. Steps 1-2 here can proceed independently.
- **#19791** (commitment pipelining): the async commitment goroutine becomes the `commitConsumer` in Step 3.
- **#19623** (IBS 2-cache rationalization): parent issue for the broader refactor.
## Out of scope
- Serial execution path (unchanged)
- The actual async commitment implementation (covered by #19791)
- BAL-driven compare-vs-write mode selection (covered by #19791)
Contributor guide
Assessment
This issue has not been assessed yet.