Skip to content

Commit b521c4e

Browse files
awskiiCopilot
andauthored
execution/commitment: streaming commitment trie (--experimental.streaming-commitment) (#21709)
**Draft / checkpoint** — streaming + parallel commitment trie, with `main` merged in and the parallel-path stability bugs fixed and validated on mainnet. ## What Adds `StreamingCommitter`, a commitment engine that overlaps trie folding with block execution (touched keys folded into per-top-nibble splits during execution; root stitched at block end). Selected via `--experimental.streaming-commitment` (precedence: streaming > parallel > concurrent). The branch stacks concurrent → parallel → streaming; streaming reuses the parallel engine's prefix-trie / split machinery and delegates `Process` to the committer. ## Stability fixes (parallel/streaming) Live mainnet validation surfaced a deterministic wrong root and two `.kv` **mmap use-after-munmap** SIGSEGVs, all now fixed: - **Wrong root** (deep storage-fold path) — the mount-only fold is the correct parallel trie; the unsound deep storage split is off by default. - **Mem-batch alias — root cause of *both* crashes.** `TemporalMemBatch.putLatest` stored values without copying; under parallel commitment they alias a `.kv` mmap of the foreground exec tx's file generation, which a background merge munmaps mid-fold. `sd.mem` is read first by every worker's `TrieContext` (shared `SharedDomains`), so a concurrent worker (`TrieContext.Branch`) or the commit flush (`SharedDomains.Commit`) reads the freed pointer → fault. Fix: **copy-on-put** so `sd.mem` owns heap bytes (`8196bac851`). (Copy-on-*get* under `latestStateLock` can't fix this — the source is already munmapped before the copy runs.) - Supporting: a **fold-scoped file-view pin** across the parallel fold (`6fadc5d9aa`), and **gating main's #21380 BranchCache off under parallel/streaming** (`1fc7e2a605`) — it gives ~nothing under parallel (which warms itself) and complicates shared-context lifetimes. ## Status - **Mainnet (live node):** parallel & streaming cross the historically-failing blocks — **25320897** (prior deterministic wrong root) and **25346499** (prior mmap fault) — with **no state-root divergence and no crash**; validation ongoing toward tip. - **Parity:** streaming root + stored branches match `ModeDirect` / `ModeParallel` across multi-depth, incremental storage collapse (partial + full delete), and whale corpora; `-race -count` clean. - `make lint` clean; `make erigon integration` builds. ## Perf (1M-whale benchmark, 18 cores) | engine | time/op | vs sequential | |--------|---------|---------------| | sequential (ModeDirect) | 1.465 s | 1.0× | | parallel | 0.410 s | 3.6× | | streaming | 0.420 s | 3.5× | ## Known follow-ups - Correct **deep storage-interior split** (depth > 64); current flat 16-way per-nibble fold already recovers the win. - Fold is **sync-bound** (goroutine coordination), not compute-bound — optimization opportunity. - Re-evaluate whether the fold-scoped pin is still needed now that copy-on-put owns the mem-batch bytes; and whether #21380's BranchCache can be made parallel-safe rather than gated. - copy-on-put adds one alloc per `DomainPut` on the exec hot path (bounded; same order as what the state/branch caches already pay). --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent ce832a4 commit b521c4e

61 files changed

Lines changed: 8345 additions & 1995 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

cmd/integration/commands/flags.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,8 @@ func withDataDir(cmd *cobra.Command) {
166166

167167
func withConcurrentCommitment(cmd *cobra.Command) {
168168
cmd.Flags().BoolVar(&statecfg.ExperimentalConcurrentCommitment, utils.ExperimentalConcurrentCommitmentFlag.Name, utils.ExperimentalConcurrentCommitmentFlag.Value, utils.ExperimentalConcurrentCommitmentFlag.Usage)
169+
cmd.Flags().BoolVar(&statecfg.ExperimentalParallelCommitment, utils.ExperimentalParallelCommitmentFlag.Name, utils.ExperimentalParallelCommitmentFlag.Value, utils.ExperimentalParallelCommitmentFlag.Usage)
170+
cmd.Flags().BoolVar(&statecfg.ExperimentalStreamingCommitment, utils.ExperimentalStreamingCommitmentFlag.Name, utils.ExperimentalStreamingCommitmentFlag.Value, utils.ExperimentalStreamingCommitmentFlag.Usage)
169171
}
170172

171173
func withBatchSize(cmd *cobra.Command) {

cmd/utils/flags.go

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1096,6 +1096,22 @@ var (
10961096
Usage: "EXPERIMENTAL: enables concurrent trie for commitment",
10971097
Value: false,
10981098
}
1099+
// ExperimentalParallelCommitmentFlag selects ParallelPatriciaHashed
1100+
// (ModeParallel) for commitment computation. Default off; flip to compare
1101+
// root hashes against a sequential sync before enabling broadly.
1102+
ExperimentalParallelCommitmentFlag = cli.BoolFlag{
1103+
Name: "experimental.parallel-commitment",
1104+
Usage: "EXPERIMENTAL: enables fully parallel trie for commitment (ParallelPatriciaHashed). Takes precedence over --experimental.concurrent-commitment if both are set.",
1105+
Value: false,
1106+
}
1107+
// ExperimentalStreamingCommitmentFlag selects the StreamingCommitter, which
1108+
// overlaps commitment fold work with block execution. Default off; takes
1109+
// precedence over the parallel/concurrent flags when set.
1110+
ExperimentalStreamingCommitmentFlag = cli.BoolFlag{
1111+
Name: "experimental.streaming-commitment",
1112+
Usage: "EXPERIMENTAL: enables streaming trie for commitment (StreamingCommitter, overlaps folding with execution). Takes precedence over --experimental.parallel-commitment and --experimental.concurrent-commitment if set.",
1113+
Value: false,
1114+
}
10991115
GDBMeFlag = cli.BoolFlag{
11001116
Name: "gdbme",
11011117
Usage: "restart erigon under gdb for debug purposes",
@@ -1962,6 +1978,14 @@ func SetEthConfig(ctx *cli.Context, nodeConfig *nodecfg.Config, cfg *ethconfig.C
19621978
cfg.ExperimentalConcurrentCommitment = true
19631979
}
19641980

1981+
if ctx.Bool(ExperimentalParallelCommitmentFlag.Name) {
1982+
cfg.ExperimentalParallelCommitment = true
1983+
}
1984+
1985+
if ctx.Bool(ExperimentalStreamingCommitmentFlag.Name) {
1986+
cfg.ExperimentalStreamingCommitment = true
1987+
}
1988+
19651989
cfg.FcuTimeout = ctx.Duration(FcuTimeoutFlag.Name)
19661990
cfg.FcuBackgroundPrune = ctx.Bool(FcuBackgroundPruneFlag.Name)
19671991
cfg.FcuBackgroundCommit = ctx.Bool(FcuBackgroundCommitFlag.Name)

common/dbg/experiments.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,10 @@ var (
4646

4747
StagesOnlyBlocks = EnvBool("STAGES_ONLY_BLOCKS", false)
4848

49+
// EnvWarmupParallelProcess gates branch-cache warmup inside the parallel
50+
// commitment Process. Off by default; opt in to prefetch under measurement.
51+
EnvWarmupParallelProcess = EnvBool("ERIGON_WARMUP_PARALLEL_PROCESS", false)
52+
4953
MdbxLockInRam = EnvBool("MDBX_LOCK_IN_RAM", false)
5054
MdbxNoSync = EnvBool("MDBX_NO_FSYNC", false)
5155
MdbxNoSyncUnsafe = EnvBool("MDBX_NO_FSYNC_UNSAFE", false)

db/integrity/commitment_integrity.go

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -210,7 +210,7 @@ func checkCommitmentRootViaFileData(ctx context.Context, tx kv.TemporalTx, br se
210210

211211
func checkCommitmentRootViaSd(ctx context.Context, tx kv.TemporalTx, f state.VisibleFile, info commitmentRootInfo, logger log.Logger) (*execctx.SharedDomains, error) {
212212
maxTxNum := f.EndRootNum() - 1
213-
sd, err := execctx.NewSharedDomains(ctx, tx, logger)
213+
sd, err := execctx.NewSharedDomains(ctx, tx, logger, execctx.WithSequentialCommitment())
214214
if err != nil {
215215
return nil, err
216216
}
@@ -1062,7 +1062,7 @@ func CheckCommitmentHistAtBlk(ctx context.Context, db kv.TemporalRoDB, br servic
10621062
return err
10631063
}
10641064
defer tx.Rollback()
1065-
sd, err := execctx.NewSharedDomains(ctx, tx, logger, execctx.WithoutDeferredBranchUpdates())
1065+
sd, err := execctx.NewSharedDomains(ctx, tx, logger, execctx.WithoutDeferredBranchUpdates(), execctx.WithSequentialCommitment())
10661066
if err != nil {
10671067
return err
10681068
}
@@ -1130,6 +1130,11 @@ func CheckCommitmentHistAtBlkRange(ctx context.Context, sc SamplerCfg, db kv.Tem
11301130
return err
11311131
}
11321132
defer tx.Rollback()
1133+
sd, err := execctx.NewSharedDomains(wCtx, tx, logger, execctx.WithoutDeferredBranchUpdates(), execctx.WithSequentialCommitment())
1134+
if err != nil {
1135+
return err
1136+
}
1137+
defer sd.Close()
11331138
idx, err := NewChangedKeysPerBlockIdx(wCtx, tx, br, windowStart, windowEnd, logger)
11341139
if err != nil {
11351140
return fmt.Errorf("CheckCommitmentHistAtBlkRange: build index window=[%d,%d): %w", windowStart, windowEnd, err)
Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
1+
// Copyright 2026 The Erigon Authors
2+
// This file is part of Erigon.
3+
//
4+
// Erigon is free software: you can redistribute it and/or modify
5+
// it under the terms of the GNU Lesser General Public License as published by
6+
// the Free Software Foundation, either version 3 of the License, or
7+
// (at your option) any later version.
8+
//
9+
// Erigon is distributed in the hope that it will be useful,
10+
// but WITHOUT ANY WARRANTY; without even the implied warranty of
11+
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12+
// GNU Lesser General Public License for more details.
13+
//
14+
// You should have received a copy of the GNU Lesser General Public License
15+
// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
16+
17+
package execctx_test
18+
19+
import (
20+
"testing"
21+
22+
"github.com/holiman/uint256"
23+
"github.com/stretchr/testify/require"
24+
25+
"github.com/erigontech/erigon/common/length"
26+
"github.com/erigontech/erigon/common/log/v3"
27+
"github.com/erigontech/erigon/db/kv"
28+
"github.com/erigontech/erigon/db/state/execctx"
29+
"github.com/erigontech/erigon/db/state/statecfg"
30+
"github.com/erigontech/erigon/execution/commitment"
31+
"github.com/erigontech/erigon/execution/types/accounts"
32+
)
33+
34+
func withCommitmentFlag(t *testing.T, variant commitment.TrieVariant) {
35+
t.Helper()
36+
origStream := statecfg.ExperimentalStreamingCommitment
37+
origPar := statecfg.ExperimentalParallelCommitment
38+
origConc := statecfg.ExperimentalConcurrentCommitment
39+
t.Cleanup(func() {
40+
statecfg.ExperimentalStreamingCommitment = origStream
41+
statecfg.ExperimentalParallelCommitment = origPar
42+
statecfg.ExperimentalConcurrentCommitment = origConc
43+
})
44+
statecfg.ExperimentalStreamingCommitment = variant == commitment.VariantStreamingHexPatricia
45+
statecfg.ExperimentalParallelCommitment = variant == commitment.VariantParallelHexPatricia
46+
statecfg.ExperimentalConcurrentCommitment = false
47+
}
48+
49+
// Update set is identical across calls so the returned roots can be compared.
50+
func runWriteCommitBatch(t *testing.T, sd *execctx.SharedDomains, rwTx kv.TemporalRwTx) []byte {
51+
t.Helper()
52+
53+
ctx := t.Context()
54+
addr := make([]byte, length.Addr)
55+
addr[0] = 0x42
56+
addr[length.Addr-1] = 0x99
57+
58+
acc := accounts.Account{
59+
Nonce: 7,
60+
Balance: *uint256.NewInt(0xdeadbeef),
61+
CodeHash: accounts.EmptyCodeHash,
62+
}
63+
pv, _, err := sd.GetLatest(kv.AccountsDomain, rwTx, addr)
64+
require.NoError(t, err)
65+
require.NoError(t, sd.DomainPut(kv.AccountsDomain, rwTx, addr, accounts.SerialiseV3(&acc), 1, pv))
66+
67+
rh, err := sd.ComputeCommitment(ctx, rwTx, false, 0, 1, "", nil)
68+
require.NoError(t, err)
69+
require.NotEmpty(t, rh)
70+
return rh
71+
}
72+
73+
func TestSharedDomains_ParallelFlag_RootEquivalence(t *testing.T) {
74+
if testing.Short() {
75+
t.Skip()
76+
}
77+
// No t.Parallel: mutates process-global statecfg flags.
78+
79+
stepSize := uint64(16)
80+
81+
runOnce := func(t *testing.T, parallel bool) []byte {
82+
t.Helper()
83+
variant := commitment.VariantHexPatriciaTrie
84+
if parallel {
85+
variant = commitment.VariantParallelHexPatricia
86+
}
87+
withCommitmentFlag(t, variant)
88+
89+
db := newTestDb(t, stepSize)
90+
91+
ctx := t.Context()
92+
rwTx, err := db.BeginTemporalRw(ctx)
93+
require.NoError(t, err)
94+
defer rwTx.Rollback()
95+
96+
sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
97+
require.NoError(t, err)
98+
defer sd.Close()
99+
100+
sd.EnableParaTrieDB(db)
101+
102+
got := sd.GetCommitmentCtx().Trie().Variant()
103+
require.Equalf(t, variant, got, "trie variant for parallel=%v", parallel)
104+
105+
return runWriteCommitBatch(t, sd, rwTx)
106+
}
107+
108+
seqRoot := runOnce(t, false)
109+
parRoot := runOnce(t, true)
110+
111+
require.Equalf(t, seqRoot, parRoot,
112+
"sequential and parallel commitment roots must match: sequential=%x parallel=%x",
113+
seqRoot, parRoot)
114+
}
115+
116+
func TestPickTrieVariant_StreamingFlag(t *testing.T) {
117+
// No t.Parallel: mutates process-global statecfg flags.
118+
withCommitmentFlag(t, commitment.VariantStreamingHexPatricia)
119+
require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
120+
121+
statecfg.ExperimentalParallelCommitment = true
122+
statecfg.ExperimentalConcurrentCommitment = true
123+
require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
124+
125+
statecfg.ExperimentalStreamingCommitment = false
126+
require.Equal(t, commitment.VariantParallelHexPatricia, execctx.PickTrieVariant())
127+
}
128+
129+
func TestSharedDomains_StreamingFlag_RootEquivalence(t *testing.T) {
130+
if testing.Short() {
131+
t.Skip()
132+
}
133+
// No t.Parallel: mutates process-global statecfg flags.
134+
135+
stepSize := uint64(16)
136+
137+
runOnce := func(t *testing.T, streaming bool) []byte {
138+
t.Helper()
139+
variant := commitment.VariantHexPatriciaTrie
140+
if streaming {
141+
variant = commitment.VariantStreamingHexPatricia
142+
}
143+
withCommitmentFlag(t, variant)
144+
145+
db := newTestDb(t, stepSize)
146+
147+
ctx := t.Context()
148+
rwTx, err := db.BeginTemporalRw(ctx)
149+
require.NoError(t, err)
150+
defer rwTx.Rollback()
151+
152+
sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
153+
require.NoError(t, err)
154+
defer sd.Close()
155+
156+
sd.EnableParaTrieDB(db)
157+
158+
got := sd.GetCommitmentCtx().Trie().Variant()
159+
require.Equalf(t, variant, got, "trie variant for streaming=%v", streaming)
160+
161+
return runWriteCommitBatch(t, sd, rwTx)
162+
}
163+
164+
seqRoot := runOnce(t, false)
165+
strRoot := runOnce(t, true)
166+
167+
require.Equalf(t, seqRoot, strRoot,
168+
"sequential and streaming commitment roots must match: sequential=%x streaming=%x",
169+
seqRoot, strRoot)
170+
}

db/state/execctx/domain_shared.go

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -132,13 +132,21 @@ type SharedDomains struct {
132132
}
133133

134134
// PickTrieVariant returns the commitment trie variant selected by the
135-
// process-wide statecfg.ExperimentalConcurrentCommitment flag. Callers that
135+
// process-wide statecfg experimental-commitment flags. Callers that
136136
// build a commitment.TrieConfig inline (e.g. short-lived RPC/builder/integrity
137-
// SharedDomains) should use this so the flag is honored consistently across
137+
// SharedDomains) should use this so the flags are honored consistently across
138138
// entry points instead of leaving Variant unset and relying on an implicit
139139
// fallback inside the trie constructor.
140140
func PickTrieVariant() commitment.TrieVariant {
141-
if statecfg.ExperimentalConcurrentCommitment {
141+
switch {
142+
// Selecting more than one experimental-commitment flag is a misconfiguration;
143+
// they are alternative paths. Streaming overlaps folding with execution, so it
144+
// wins over parallel, which in turn parallelizes deeper than concurrent.
145+
case statecfg.ExperimentalStreamingCommitment:
146+
return commitment.VariantStreamingHexPatricia
147+
case statecfg.ExperimentalParallelCommitment:
148+
return commitment.VariantParallelHexPatricia
149+
case statecfg.ExperimentalConcurrentCommitment:
142150
return commitment.VariantConcurrentHexPatricia
143151
}
144152
return commitment.VariantHexPatriciaTrie

db/state/execctx/options.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,3 +34,10 @@ func WithTrieConfig(cfg commitment.TrieConfig) SharedDomainOption {
3434
func WithoutDeferredBranchUpdates() SharedDomainOption {
3535
return func(o *sharedDomainOptions) { o.trieCfg.DeferBranchUpdates = false }
3636
}
37+
38+
// WithSequentialCommitment forces the sequential HexPatriciaHashed trie regardless
39+
// of the experimental parallel/concurrent flags — for one-shot / empty-DB paths
40+
// (e.g. genesis) that wire no trie-context factory for the parallel trie.
41+
func WithSequentialCommitment() SharedDomainOption {
42+
return func(o *sharedDomainOptions) { o.trieCfg.Variant = commitment.VariantHexPatriciaTrie }
43+
}

db/state/squeeze.go

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -951,9 +951,16 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
951951
}
952952
roTx.Rollback()
953953

954+
streaming := statecfg.ExperimentalStreamingCommitment
955+
parallel := statecfg.ExperimentalParallelCommitment
954956
concurrent := statecfg.ExperimentalConcurrentCommitment
955957
trieVariant := commitment.VariantHexPatriciaTrie
956-
if concurrent {
958+
switch {
959+
case streaming:
960+
trieVariant = commitment.VariantStreamingHexPatricia
961+
case parallel:
962+
trieVariant = commitment.VariantParallelHexPatricia
963+
case concurrent:
957964
trieVariant = commitment.VariantConcurrentHexPatricia
958965
}
959966

@@ -990,7 +997,7 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
990997
domains.SetTxNum(lastTxnumInShard - 1)
991998
currentTxNum := lastTxnumInShard - 1
992999
domains.GetCommitmentCtx().SetStateReader(commitmentdb.NewFilesOnlyStateReader(rwTx, lastTxnumInShard-1))
993-
if concurrent {
1000+
if concurrent || parallel || streaming {
9941001
domains.EnableParaTrieDB(rwDb)
9951002
}
9961003

db/state/statecfg/state_schema.go

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,17 @@ func (s *SchemaGen) GetBlockIdxFilesCfg(name string) BlockIdxFilesCfg {
189189

190190
var ExperimentalConcurrentCommitment = false // set true to use concurrent commitment by default
191191

192+
// ExperimentalParallelCommitment toggles the ParallelPatriciaHashed trie path
193+
// (commitment.ModeParallel + VariantParallelHexPatricia). Default false.
194+
// Takes precedence over ExperimentalConcurrentCommitment when both are set.
195+
var ExperimentalParallelCommitment = false
196+
197+
// ExperimentalStreamingCommitment toggles the StreamingCommitter trie path
198+
// (commitment.ModeParallel + VariantStreamingHexPatricia), which overlaps
199+
// commitment folding with execution. Default false. Takes precedence over
200+
// ExperimentalParallelCommitment and ExperimentalConcurrentCommitment.
201+
var ExperimentalStreamingCommitment = false
202+
192203
var Schema = SchemaGen{
193204
AccountsDomain: DomainCfg{
194205
Name: kv.AccountsDomain, ValuesTable: kv.TblAccountVals,

db/state/temporal_mem_batch.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -173,7 +173,9 @@ func (sd *TemporalMemBatch) putLatest(domain kv.Domain, key string, val []byte,
173173
}
174174
}
175175

176-
valWithStep := dataWithTxNum{data: val, txNum: txNum}
176+
// Own the bytes now: val may alias a .kv mmap (the foreground exec tx's file generation)
177+
// that a background merge can munmap while a concurrent commitment worker reads sd.mem.
178+
valWithStep := dataWithTxNum{data: common.Copy(val), txNum: txNum}
177179
putKeySize := 0
178180
putValueSize := 0
179181
if domain == kv.StorageDomain {

0 commit comments

Comments
 (0)