diff --git a/cmd/evm/staterunner_test.go b/cmd/evm/staterunner_test.go
index b1a8b4cd36b..87c21b71f4f 100644
--- a/cmd/evm/staterunner_test.go
+++ b/cmd/evm/staterunner_test.go
@@ -35,10 +35,8 @@ import (
func TestNewStateTestSharedDomainsUsesSelectedCommitment(t *testing.T) {
originalParallel := statecfg.ExperimentalParallelCommitment
- originalStreaming := statecfg.ExperimentalStreamingCommitment
t.Cleanup(func() {
statecfg.ExperimentalParallelCommitment = originalParallel
- statecfg.ExperimentalStreamingCommitment = originalStreaming
})
for _, tc := range []struct {
@@ -51,7 +49,6 @@ func TestNewStateTestSharedDomainsUsesSelectedCommitment(t *testing.T) {
} {
t.Run(tc.name, func(t *testing.T) {
statecfg.ExperimentalParallelCommitment = tc.parallel
- statecfg.ExperimentalStreamingCommitment = false
db, tx := temporaltest.NewTestTx(t)
sd, err := newStateTestSharedDomains(db, tx)
diff --git a/cmd/integration/commands/flags.go b/cmd/integration/commands/flags.go
index d5551a57689..eb73edb209f 100644
--- a/cmd/integration/commands/flags.go
+++ b/cmd/integration/commands/flags.go
@@ -175,7 +175,6 @@ func withDataDir(cmd *cobra.Command) {
func withExperimentalCommitment(cmd *cobra.Command) {
cmd.Flags().BoolVar(&statecfg.ExperimentalParallelCommitment, utils.ExperimentalParallelCommitmentFlag.Name, statecfg.ExperimentalParallelCommitment, utils.ExperimentalParallelCommitmentFlag.Usage)
- cmd.Flags().BoolVar(&statecfg.ExperimentalStreamingCommitment, utils.ExperimentalStreamingCommitmentFlag.Name, statecfg.ExperimentalStreamingCommitment, utils.ExperimentalStreamingCommitmentFlag.Usage)
}
func withBatchSize(cmd *cobra.Command) {
diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
index d493c1fa9a3..b86677bc38f 100644
--- a/cmd/utils/flags.go
+++ b/cmd/utils/flags.go
@@ -1166,14 +1166,6 @@ var (
Usage: "EXPERIMENTAL: enables fully parallel trie for commitment (ParallelPatriciaHashed).",
Value: false,
}
- // ExperimentalStreamingCommitmentFlag selects the StreamingCommitter, which
- // overlaps commitment fold work with block execution. Default off; takes
- // precedence over the parallel flag when set.
- ExperimentalStreamingCommitmentFlag = cli.BoolFlag{
- Name: "experimental.streaming-commitment",
- Usage: "EXPERIMENTAL: enables streaming trie for commitment (StreamingCommitter, overlaps folding with execution). Takes precedence over --experimental.parallel-commitment if set.",
- Value: false,
- }
GDBMeFlag = cli.BoolFlag{
Name: "gdbme",
Usage: "restart erigon under gdb for debug purposes",
@@ -2060,10 +2052,6 @@ func SetEthConfig(nodeCtx context.Context, ctx *cli.Command, nodeConfig *nodecfg
cfg.ExperimentalParallelCommitment = true
}
- if ctx.Bool(ExperimentalStreamingCommitmentFlag.Name) {
- cfg.ExperimentalStreamingCommitment = true
- }
-
cfg.FcuTimeout = ctx.Duration(FcuTimeoutFlag.Name)
cfg.FcuBackgroundPrune = ctx.Bool(FcuBackgroundPruneFlag.Name)
diff --git a/db/state/execctx/commitment_flag_test.go b/db/state/execctx/commitment_flag_test.go
index ef11b126c63..241e913b97f 100644
--- a/db/state/execctx/commitment_flag_test.go
+++ b/db/state/execctx/commitment_flag_test.go
@@ -33,13 +33,10 @@ import (
func withCommitmentFlag(t *testing.T, variant commitment.TrieVariant) {
t.Helper()
- origStream := statecfg.ExperimentalStreamingCommitment
origPar := statecfg.ExperimentalParallelCommitment
t.Cleanup(func() {
- statecfg.ExperimentalStreamingCommitment = origStream
statecfg.ExperimentalParallelCommitment = origPar
})
- statecfg.ExperimentalStreamingCommitment = variant == commitment.VariantStreamingHexPatricia
statecfg.ExperimentalParallelCommitment = variant == commitment.VariantParallelHexPatricia
}
@@ -109,58 +106,3 @@ func TestSharedDomains_ParallelFlag_RootEquivalence(t *testing.T) {
"sequential and parallel commitment roots must match: sequential=%x parallel=%x",
seqRoot, parRoot)
}
-
-func TestPickTrieVariant_StreamingFlag(t *testing.T) {
- // No t.Parallel: mutates process-global statecfg flags.
- withCommitmentFlag(t, commitment.VariantStreamingHexPatricia)
- require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
-
- statecfg.ExperimentalParallelCommitment = true
- require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
-
- statecfg.ExperimentalStreamingCommitment = false
- require.Equal(t, commitment.VariantParallelHexPatricia, execctx.PickTrieVariant())
-}
-
-func TestSharedDomains_StreamingFlag_RootEquivalence(t *testing.T) {
- if testing.Short() {
- t.Skip()
- }
- // No t.Parallel: mutates process-global statecfg flags.
-
- stepSize := uint64(16)
-
- runOnce := func(t *testing.T, streaming bool) []byte {
- t.Helper()
- variant := commitment.VariantHexPatriciaTrie
- if streaming {
- variant = commitment.VariantStreamingHexPatricia
- }
- withCommitmentFlag(t, variant)
-
- db := newTestDb(t, stepSize)
-
- ctx := t.Context()
- rwTx, err := db.BeginTemporalRw(ctx)
- require.NoError(t, err)
- defer rwTx.Rollback()
-
- sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
- require.NoError(t, err)
- defer sd.Close()
-
- sd.EnableParaTrieDB(db)
-
- got := sd.GetCommitmentCtx().Trie().Variant()
- require.Equalf(t, variant, got, "trie variant for streaming=%v", streaming)
-
- return runWriteCommitBatch(t, sd, rwTx)
- }
-
- seqRoot := runOnce(t, false)
- strRoot := runOnce(t, true)
-
- require.Equalf(t, seqRoot, strRoot,
- "sequential and streaming commitment roots must match: sequential=%x streaming=%x",
- seqRoot, strRoot)
-}
diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go
index be5e8d8f3e9..071176eb652 100644
--- a/db/state/execctx/domain_shared.go
+++ b/db/state/execctx/domain_shared.go
@@ -290,13 +290,7 @@ type SharedDomains struct {
// entry points instead of leaving Variant unset and relying on an implicit
// fallback inside the trie constructor.
func PickTrieVariant() commitment.TrieVariant {
- switch {
- // Selecting more than one experimental-commitment flag is a misconfiguration;
- // they are alternative paths. Streaming overlaps folding with execution, so it
- // wins over parallel.
- case statecfg.ExperimentalStreamingCommitment:
- return commitment.VariantStreamingHexPatricia
- case statecfg.ExperimentalParallelCommitment:
+ if statecfg.ExperimentalParallelCommitment {
return commitment.VariantParallelHexPatricia
}
return commitment.VariantHexPatriciaTrie
diff --git a/db/state/squeeze.go b/db/state/squeeze.go
index 5fc49944dd2..17df4b4ef9b 100644
--- a/db/state/squeeze.go
+++ b/db/state/squeeze.go
@@ -1020,13 +1020,8 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
}
roTx.Rollback()
- streaming := statecfg.ExperimentalStreamingCommitment
- parallel := statecfg.ExperimentalParallelCommitment
trieVariant := commitment.VariantHexPatriciaTrie
- switch {
- case streaming:
- trieVariant = commitment.VariantStreamingHexPatricia
- case parallel:
+ if statecfg.ExperimentalParallelCommitment {
trieVariant = commitment.VariantParallelHexPatricia
}
@@ -1063,7 +1058,7 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
domains.SetTxNum(lastTxnumInShard - 1)
currentTxNum := lastTxnumInShard - 1
domains.GetCommitmentCtx().SetStateReader(commitmentdb.NewFilesOnlyStateReader(rwTx, lastTxnumInShard-1))
- if parallel || streaming {
+ if statecfg.ExperimentalParallelCommitment {
domains.EnableParaTrieDB(rwDb)
}
diff --git a/db/state/statecfg/state_schema.go b/db/state/statecfg/state_schema.go
index 2d4ec5894af..7d1d3057a71 100644
--- a/db/state/statecfg/state_schema.go
+++ b/db/state/statecfg/state_schema.go
@@ -201,12 +201,6 @@ func commitmentKVWriteVersion(c *DomainCfg) version.Version {
// COMMITMENT_PARALLEL env var (or the CLI flag) turns it on.
var ExperimentalParallelCommitment = dbg.EnvBool("COMMITMENT_PARALLEL", false)
-// ExperimentalStreamingCommitment toggles the StreamingCommitter trie path
-// (commitment.ModeParallel + VariantStreamingHexPatricia), which overlaps
-// commitment folding with execution. Default false. Takes precedence over
-// ExperimentalParallelCommitment.
-var ExperimentalStreamingCommitment = false
-
var Schema = SchemaGen{
AccountsDomain: DomainCfg{
Name: kv.AccountsDomain, ValuesTable: kv.TblAccountVals,
diff --git a/execution/commitment/additive_updates_test.go b/execution/commitment/additive_updates_test.go
index 4be0631b171..60cff293509 100644
--- a/execution/commitment/additive_updates_test.go
+++ b/execution/commitment/additive_updates_test.go
@@ -83,25 +83,4 @@ func TestAdditiveTouch(t *testing.T) {
require.NoError(t, err)
require.Equal(t, seqRoot, parRoot, "additive partial touches must fold to the merged root")
})
-
- t.Run("streaming", func(t *testing.T) {
- keys, partials, merged := additiveCorpus()
- seqRoot, _ := sequentialRoot(t, keys, merged)
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- require.NoError(t, ms.applyPlainUpdates(keys, merged))
-
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- defer sc.Release()
- sc.SetNumWorkers(2)
- for i, k := range keys {
- sc.TouchKey(KeyToHexNibbleHash(k), k, partials[i][0])
- sc.TouchKey(KeyToHexNibbleHash(k), k, partials[i][1])
- }
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "streaming additive touches must fold to the merged root")
- })
}
diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go
index 1b1c7736747..d514716b8bc 100644
--- a/execution/commitment/commitment.go
+++ b/execution/commitment/commitment.go
@@ -139,9 +139,8 @@ type TrieVariant string
const (
// VariantHexPatriciaTrie used as default commitment approach
- VariantHexPatriciaTrie TrieVariant = "hex-patricia-hashed"
- VariantParallelHexPatricia TrieVariant = "hex-parallel-patricia-hashed"
- VariantStreamingHexPatricia TrieVariant = "hex-streaming-patricia-hashed"
+ VariantHexPatriciaTrie TrieVariant = "hex-patricia-hashed"
+ VariantParallelHexPatricia TrieVariant = "hex-parallel-patricia-hashed"
)
// InitializeTrieAndUpdates constructs the trie + updates buffer from cfg.
@@ -152,13 +151,6 @@ func InitializeTrieAndUpdates(mode Mode, tmpdir string, cfg TrieConfig) (Trie, *
trie := NewParallelPatriciaHashed(nil, length.Addr, cfg)
tree := NewUpdates(ModeParallel, tmpdir, KeyToHexNibbleHash)
return trie, tree
- case VariantStreamingHexPatricia:
- trie := NewParallelPatriciaHashed(nil, length.Addr, cfg)
- sc := NewStreamingCommitter(nil, length.Addr, cfg)
- trie.SetStreamingCommitter(sc)
- tree := NewUpdates(ModeParallel, tmpdir, KeyToHexNibbleHash)
- tree.SetStreamingCommitter(sc)
- return trie, tree
case VariantHexPatriciaTrie:
fallthrough
default:
@@ -1375,12 +1367,6 @@ const (
ModeParallel Mode = 3
)
-// streamingSink receives touched keys for a StreamingCommitter's fold; plainKey
-// and update must stay valid until the committer's Process call.
-type streamingSink interface {
- TouchKey(hashedKey, plainKey []byte, update *Update)
-}
-
func (m Mode) String() string {
switch m {
case ModeDisabled:
@@ -1422,10 +1408,6 @@ type Updates struct {
directBytes int
directMemLimit int
- // streaming (ModeParallel only) forwards every touched key to streamer.
- streaming bool
- streamer streamingSink
-
batchSlab []KeyUpdate // grow-only slab for HashSort batch (avoids per-key heap allocs)
// Ring of byte arenas for HashSort key copies; a slot is reused only after its prior generation's warm items drain.
@@ -1501,13 +1483,9 @@ func (t *Updates) hashKey(key []byte) []byte {
return t.hasher(key)
}
-// NewEmpty creates a fresh Updates matching the receiver. The streaming sink must
-// carry over, or a buffer rotated mid-stream silently computes a stale root.
+// NewEmpty creates a fresh Updates matching the receiver.
func (t *Updates) NewEmpty() *Updates {
- n := NewUpdates(t.mode, t.tmpdir, t.hasher)
- n.streamer = t.streamer
- n.streaming = t.streaming
- return n
+ return NewUpdates(t.mode, t.tmpdir, t.hasher)
}
func NewUpdates(m Mode, tmpdir string, hasher keyHasher) *Updates {
@@ -1610,15 +1588,6 @@ func (t *Updates) spillDirect() {
func (t *Updates) Mode() Mode { return t.mode }
-// SetStreamingCommitter forwards ModeParallel touches to sink; nil disables streaming.
-func (t *Updates) SetStreamingCommitter(sink streamingSink) {
- t.streamer = sink
- t.streaming = sink != nil
-}
-
-// Streaming reports whether touches are being forwarded to a StreamingCommitter.
-func (t *Updates) Streaming() bool { return t.streaming }
-
// PlainKeys returns a copy of the set of plain keys that have been touched.
// Meaningful only in ModeDirect and ModeParallel; nil otherwise.
func (t *Updates) PlainKeys() map[string]struct{} {
@@ -1666,9 +1635,8 @@ func (t *Updates) TouchPlainKey(key string, val []byte, fn func(c *KeyUpdate, va
t.keys[key] = struct{}{}
}
case ModeParallel:
- // The dedup map only guards plain-key interning: every touch reaches the prefix
- // trie and the streamer, so a same-block re-touch invalidates any eager fold of
- // its split instead of leaving it stale.
+ // The dedup map only guards plain-key interning: every touch still reaches
+ // the prefix trie, so a same-block re-touch updates its merged value there.
keyBytes := common.ToBytesZeroCopy(key)
hashedKey := t.hashKey(keyBytes)
ik := keyBytes
@@ -1677,9 +1645,6 @@ func (t *Updates) TouchPlainKey(key string, val []byte, fn func(c *KeyUpdate, va
t.keys[key] = struct{}{}
}
t.parallel.Insert(hashedKey, ik, nil)
- if t.streaming && t.streamer != nil {
- t.streamer.TouchKey(hashedKey, ik, nil)
- }
default:
}
}
@@ -1747,9 +1712,6 @@ func (t *Updates) TouchPlainKeyDirect(key string, update *Update) {
t.keys[key] = struct{}{}
}
t.parallel.Insert(hashedKey, ik, u)
- if t.streaming && t.streamer != nil {
- t.streamer.TouchKey(hashedKey, ik, u)
- }
default:
}
}
diff --git a/execution/commitment/commitmentdb/commitment_context.go b/execution/commitment/commitmentdb/commitment_context.go
index 77165a3c43b..e419d543b2f 100644
--- a/execution/commitment/commitmentdb/commitment_context.go
+++ b/execution/commitment/commitmentdb/commitment_context.go
@@ -57,7 +57,7 @@ type SharedDomainsCommitmentContext struct {
sharedDomains sd
updates *commitment.Updates
patriciaTrie commitment.Trie
- variant commitment.TrieVariant // selected trie engine, for the [commitment] log (updates.Mode() is ModeParallel for both parallel and streaming)
+ variant commitment.TrieVariant // selected trie engine, for the [commitment] log (updates.Mode() is ModeParallel for the parallel trie)
justRestored atomic.Bool // set to true when commitment trie was just restored from snapshot
traceW io.Writer
stateReader StateReader
@@ -75,8 +75,8 @@ type SharedDomainsCommitmentContext struct {
// pendingUpdate stores a single deferred branch update to be flushed at the next ComputeCommitment call.
pendingUpdate *commitment.PendingCommitmentUpdate
- // pendingVariant holds a parallel/streaming trie selection that waits for
- // EnableParaTrieDB: those variants need the DB-backed TrieContextFactory.
+ // pendingVariant holds a parallel trie selection that waits for
+ // EnableParaTrieDB: that variant needs the DB-backed TrieContextFactory.
pendingVariant commitment.TrieVariant
pendingCfg commitment.TrieConfig
}
@@ -231,12 +231,11 @@ func NewSharedDomainsCommitmentContext(sd sd, mode commitment.Mode, tmpDir strin
NumWorkers: cfg.WarmupNumWorkersOrDefault(),
},
}
- // The parallel and streaming tries need a per-worker TrieContextFactory that
- // only DB-backed consumers can provide (via EnableParaTrieDB). Start on the
- // sequential trie and upgrade when the DB arrives, so context holders that
- // never wire one (RPC, integrity, tests) keep working under a global variant
- // selection.
- if variant == commitment.VariantParallelHexPatricia || variant == commitment.VariantStreamingHexPatricia {
+ // The parallel trie needs a per-worker TrieContextFactory that only DB-backed
+ // consumers can provide (via EnableParaTrieDB). Start on the sequential trie
+ // and upgrade when the DB arrives, so context holders that never wire one
+ // (RPC, integrity, tests) keep working under a global variant selection.
+ if variant == commitment.VariantParallelHexPatricia {
ctx.pendingVariant = variant
cfg.Variant = commitment.VariantHexPatriciaTrie
ctx.pendingCfg = cfg
@@ -561,8 +560,7 @@ func (sdc *SharedDomainsCommitmentContext) ComputeCommitment(ctx context.Context
trie.SetTrieContextFactory(concurrentFactory)
default:
// Serial: this factory only serves page-cache warmup, which does not
- // compute the root, so its reads need no generation pin. (Streaming is
- // a *ParallelPatriciaHashed and takes the pinned branch above.)
+ // compute the root, so its reads need no generation pin.
warmupConfig.CtxFactory = sdc.warmupTrieContextFactory(sdc.paraTrieDB, txNum)
}
}
@@ -792,7 +790,7 @@ func DecodeTxBlockNums(v []byte) (txNum, blockNum uint64) {
// Found value does not become current state.
func (sdc *SharedDomainsCommitmentContext) LatestCommitmentState(trieContext *TrieContext) (blockNum, txNum uint64, state []byte, err error) {
tv := sdc.patriciaTrie.Variant()
- if tv != commitment.VariantHexPatriciaTrie && tv != commitment.VariantParallelHexPatricia && tv != commitment.VariantStreamingHexPatricia {
+ if tv != commitment.VariantHexPatriciaTrie && tv != commitment.VariantParallelHexPatricia {
return 0, 0, nil, errors.New("state storing is only supported hex patricia trie")
}
var step kv.Step
@@ -931,7 +929,7 @@ func (sdc *SharedDomainsCommitmentContext) restorePatriciaState(value []byte) (u
return 0, 0, errors.New("cannot typecast hex patricia trie")
}
}
- if tv == commitment.VariantParallelHexPatricia || tv == commitment.VariantStreamingHexPatricia {
+ if tv == commitment.VariantParallelHexPatricia {
var ok bool
ppht, ok = sdc.patriciaTrie.(*commitment.ParallelPatriciaHashed)
if !ok {
diff --git a/execution/commitment/deepfold_test.go b/execution/commitment/deepfold_test.go
index 270b03f8d79..04397ffb42b 100644
--- a/execution/commitment/deepfold_test.go
+++ b/execution/commitment/deepfold_test.go
@@ -227,8 +227,6 @@ func TestStreaming_ExtensionToppedMountSplit(t *testing.T) {
mode runMode
}{
{"parallel", modeParallel},
- {"streaming", modeStreaming},
- {"streaming_scheduled", modeStreamingScheduled},
} {
for _, w := range []int{1, 4, 8} {
roots, ms := runEngineBatches(t, tc.mode, w, batches)
@@ -399,7 +397,7 @@ func TestDeepFold_PreExistingWhale_SingleNibbleOnDisk(t *testing.T) {
// A FRESH whale — its account absent from the pre-state trie — provably has nothing on
// disk beneath its storage prefix, so the deep fold seeds an empty base and folds the
-// slots concurrently instead of demoting to serial streaming.
+// slots concurrently instead of demoting to serial recursion.
func TestDeepFold_FreshWhaleFoldsParallel(t *testing.T) {
k1, u1, _, _ := buildSubsetTouchedWhale(20260707, nibs(3, 7), nil, 700, 0)
fk, fu := buildMixedCorpus(555, 200)
@@ -410,17 +408,9 @@ func TestDeepFold_FreshWhaleFoldsParallel(t *testing.T) {
ms := NewMockState(t)
ms.SetConcurrentCommitment(true)
- require.NoError(t, ms.applyPlainUpdates(keys, upds))
- sc := newStreamCommitter(t, ms, 4, false)
- defer sc.Release()
- touchAll(sc, keys)
- got, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, got, "fresh-whale concurrent fold diverged from sequential")
- require.Positive(t, sc.DeepLocalFolds(), "a fresh whale must take the concurrent deep fold, not the serial demotion")
-
- parRoot, _ := engineRoot(t, modeParallel, 4, keys, upds)
+ parRoot, _, deepFolds := parallelBatchDeepFolds(t, ms, 4, keys, upds, nil)
require.Equal(t, seqRoot, parRoot)
+ require.Positive(t, deepFolds, "a fresh whale must take the concurrent deep fold, not the serial demotion")
}
// The demotion gate stays for accounts present in the pre-state without a branch record
@@ -435,18 +425,10 @@ func TestDeepFold_ExistingWhaleStillDemotes(t *testing.T) {
ms := NewMockState(t)
ms.SetConcurrentCommitment(true)
- sc := newStreamCommitter(t, ms, 4, false)
- defer sc.Release()
- require.NoError(t, ms.applyPlainUpdates(k1, u1))
- touchAll(sc, k1)
- _, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- touchAll(sc, k2)
- got, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, got)
- require.Zero(t, sc.DeepLocalFolds(), "an account present in the pre-state must keep the serial demotion")
+ _, blob, _ := parallelBatchDeepFolds(t, ms, 4, k1, u1, nil)
+ parRoot, _, deepFolds := parallelBatchDeepFolds(t, ms, 4, k2, u2, blob)
+ require.Equal(t, seqRoot, parRoot)
+ require.Zero(t, deepFolds, "an account present in the pre-state must keep the serial demotion")
}
// A whale's storage collapses to a single surviving slot through the streaming recursion (sub-threshold
@@ -529,8 +511,6 @@ func TestDeepFold_SingleSlotCollapseThenDeepReexpand(t *testing.T) {
mode runMode
}{
{"parallel", modeParallel},
- {"streaming", modeStreaming},
- {"streaming_scheduled", modeStreamingScheduled},
} {
for _, w := range []int{1, 4, 8} {
roots, ms := runEngineBatches(t, tc.mode, w, batches)
@@ -605,8 +585,6 @@ func TestDeepFold_SurvivorCollapseThenRetouch(t *testing.T) {
mode runMode
}{
{"parallel", modeParallel},
- {"streaming", modeStreaming},
- {"streaming_scheduled", modeStreamingScheduled},
} {
for _, w := range []int{1, 4, 8} {
roots, ms := runEngineBatches(t, tc.mode, w, batches)
@@ -677,8 +655,6 @@ func TestDeepFold_EmptyStorageThenRepopulate(t *testing.T) {
mode runMode
}{
{"parallel", modeParallel},
- {"streaming", modeStreaming},
- {"streaming_scheduled", modeStreamingScheduled},
} {
for _, w := range []int{1, 4, 8} {
roots, ms := runEngineBatches(t, tc.mode, w, batches)
@@ -978,3 +954,82 @@ func TestFillFromLowerCell_StorageBranchSyncsNavPath(t *testing.T) {
require.Equal(t, []byte{0x5, 0xd}, branchCell.hashedExtension[:branchCell.hashedExtLen],
"a keyless cell deep in storage still navigates by its extension")
}
+
+// keyArena backs collectSubtreeKeys in streaming_deep_fold.go.
+func TestKeyArena_PointerStability(t *testing.T) {
+ var arena keyArena
+
+ inputs := make([][]byte, 0, 4096)
+ got := make([][]byte, 0, 4096)
+ // 4096 small keys roll the arena over at least two chunks.
+ for i := range 4096 {
+ in := bytes.Repeat([]byte{byte(i), byte(i >> 8)}, 32)
+ inputs = append(inputs, in)
+ got = append(got, arena.copy(in))
+ }
+ // Oversized key forces the max(keyArenaChunk, len) allocation path.
+ big := bytes.Repeat([]byte{0xAB}, keyArenaChunk+128)
+ inputs = append(inputs, big)
+ got = append(got, arena.copy(big))
+
+ for i, in := range inputs {
+ require.True(t, bytes.Equal(in, got[i]),
+ "key %d corrupted: returned slice does not equal its input", i)
+ require.Equal(t, len(got[i]), cap(got[i]),
+ "key %d not full-cap: a caller append could overwrite the next key", i)
+ }
+
+ for i := range got {
+ for j := range got[i] {
+ got[i][j] = byte(i)
+ }
+ }
+ for i := range got {
+ for j := range got[i] {
+ require.Equal(t, byte(i), got[i][j],
+ "key %d overlaps another arena slice (overwritten at byte %d)", i, j)
+ }
+ }
+}
+
+func TestKeyArena_ChunkSizedFromRemaining(t *testing.T) {
+ const keyLen = 144
+
+ t.Run("small subtree does not burn a full chunk", func(t *testing.T) {
+ const keys = 32
+ arena := keyArena{remaining: keys}
+ for range keys {
+ arena.copy(make([]byte, keyLen))
+ }
+ require.Equal(t, keys*keyLen, cap(arena.buf))
+ })
+
+ t.Run("large subtree still caps at one chunk", func(t *testing.T) {
+ arena := keyArena{remaining: 10 * keyArenaChunk / keyLen}
+ arena.copy(make([]byte, keyLen))
+ require.Equal(t, keyArenaChunk, cap(arena.buf))
+ })
+
+ t.Run("oversized key gets its own backing", func(t *testing.T) {
+ arena := keyArena{remaining: 4}
+ got := arena.copy(make([]byte, 2*keyArenaChunk))
+ require.Len(t, got, 2*keyArenaChunk)
+ require.Equal(t, 2*keyArenaChunk, cap(arena.buf))
+ })
+
+ t.Run("copies stay stable across a chunk swap", func(t *testing.T) {
+ arena := keyArena{remaining: 2}
+ first := arena.copy(bytes.Repeat([]byte{0xAA}, keyLen))
+ for i := range 8 {
+ arena.copy(bytes.Repeat([]byte{byte(i)}, keyLen))
+ }
+ require.Equal(t, bytes.Repeat([]byte{0xAA}, keyLen), first)
+ })
+
+ t.Run("unhinted arena falls back to one key per chunk growth", func(t *testing.T) {
+ var arena keyArena
+ got := arena.copy(make([]byte, keyLen))
+ require.Len(t, got, keyLen)
+ require.Equal(t, keyLen, cap(arena.buf))
+ })
+}
diff --git a/execution/commitment/mode_parallel_lifecycle_test.go b/execution/commitment/mode_parallel_lifecycle_test.go
index a8720200edb..38b55c0d68c 100644
--- a/execution/commitment/mode_parallel_lifecycle_test.go
+++ b/execution/commitment/mode_parallel_lifecycle_test.go
@@ -66,55 +66,6 @@ func touchBatch(t *testing.T, ms *MockState, ut *Updates, keys [][]byte, upds []
}
}
-type countingSink struct{ calls int }
-
-func (c *countingSink) TouchKey(hashedKey, plainKey []byte, update *Update) { c.calls++ }
-
-// The dedup map only dedups plain-key interning: every touch — including a repeat of an
-// already-collected key — must be forwarded to the streamer, or a scheduler's eagerly
-// folded split goes stale within the block.
-func TestModeParallel_RetouchReachesStreamer(t *testing.T) {
- t.Parallel()
- ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
- defer ut.Close()
- sink := &countingSink{}
- ut.SetStreamingCommitter(sink)
-
- a := findAddressForNibble(3, 999)
- ut.TouchPlainKey(string(a), nil, ut.TouchAccount)
- ut.TouchPlainKey(string(a), nil, ut.TouchAccount)
- require.Equal(t, 2, sink.calls, "a re-touch must be forwarded to the streamer, not deduped")
- require.Equal(t, uint64(1), ut.Size(), "interning stays deduped")
-}
-
-// A node carries one ModeParallel Updates buffer across blocks: a key re-touched in a
-// later block must land in that block's fold instead of being dropped by stale per-buffer
-// dedup state. Pins the carried-buffer lifecycle end to end.
-func TestModeParallel_StreamingRetouchAcrossBlocks(t *testing.T) {
- t.Parallel()
- k1, u1, k2, u2, kc, uc := lifecycleCorpus()
- oracle, _ := engineRoot(t, modeSeq, 0, kc, uc)
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut.Close()
- defer trie.Release()
- pt := trie.(*ParallelPatriciaHashed)
- pt.SetNumWorkers(4)
- pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt.ResetContext(ms)
-
- touchBatch(t, ms, ut, k1, u1)
- processRoot(t, trie, ut)
-
- touchBatch(t, ms, ut, k2, u2)
- got := processRoot(t, trie, ut)
- require.Equal(t, oracle, got, "re-touched keys were dropped by the stale dedup map")
-}
-
// Process must consume the ModeParallel collection the way HashSort consumes
// ModeDirect/ModeUpdate: a carried Updates buffer starts every block empty, so block N+1
// folds only its own touches instead of the union of everything since batch start.
@@ -150,38 +101,6 @@ func TestModeParallel_ProcessConsumesUpdates(t *testing.T) {
again := processRoot(t, tr, ut)
require.Equal(t, got, again, "a zero-touch Process must return the carried root")
})
-
- t.Run("streaming", func(t *testing.T) {
- t.Parallel()
- oracle, _ := engineRoot(t, modeSeq, 0, kc, uc)
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut.Close()
- defer trie.Release()
- pt := trie.(*ParallelPatriciaHashed)
- pt.SetNumWorkers(4)
- pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt.ResetContext(ms)
-
- touchBatch(t, ms, ut, k1, u1)
- processRoot(t, trie, ut)
- require.Zero(t, ut.Size(), "streaming Process left the touched-key collection unconsumed")
- if root := ut.parallel.trie.root; root != nil {
- require.Zero(t, root.subtreeCount, "streaming Process left the dual-inserted prefix trie populated")
- }
-
- touchBatch(t, ms, ut, k2, u2)
- got := processRoot(t, trie, ut)
- require.Zero(t, ut.Size())
- require.Equal(t, oracle, got)
-
- again := processRoot(t, trie, ut)
- require.Equal(t, got, again, "a zero-touch streaming Process must return the carried root, not the empty root")
- })
}
// A failed Process must leave the collection intact so the caller's retry folds the
@@ -212,28 +131,4 @@ func TestModeParallel_ErrorKeepsCollection(t *testing.T) {
require.Zero(t, ut.Size())
require.Equal(t, oracle, got)
})
-
- t.Run("streaming", func(t *testing.T) {
- t.Parallel()
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut.Close()
- defer trie.Release()
- pt := trie.(*ParallelPatriciaHashed)
- pt.SetNumWorkers(4)
- pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt.ResetContext(ms)
-
- touchBatch(t, ms, ut, k1, u1)
- _, err := trie.Process(canceled, ut, "", nil, WarmupConfig{})
- require.Error(t, err)
- require.Equal(t, uint64(len(k1)), ut.Size(), "error path must keep the collection for the retry")
-
- got := processRoot(t, trie, ut)
- require.Zero(t, ut.Size())
- require.Equal(t, oracle, got)
- })
}
diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go
index 422ff6471bb..d29ccd74294 100644
--- a/execution/commitment/parallel_mount.go
+++ b/execution/commitment/parallel_mount.go
@@ -74,6 +74,25 @@ func (hph *HexPatriciaHashed) mountTo(root *HexPatriciaHashed, nibble int) {
copy(hph.grid[:n], root.grid[:n])
}
+// stitchSplitCells drops each folded split cell into the base row at its top-nibble slot;
+// foldMounted already returns cells excluding the mount nibble, so they are stitched verbatim.
+func stitchSplitCells(base *HexPatriciaHashed, cells *[16]cell, present *[16]bool) {
+ for nib := range 16 {
+ if !present[nib] {
+ continue
+ }
+ c := cells[nib]
+ base.touchMap[0] |= uint16(1) << nib
+ if !c.IsEmpty() {
+ base.afterMap[0] |= uint16(1) << nib
+ } else {
+ base.afterMap[0] &^= uint16(1) << nib
+ }
+ base.depths[0] = 1
+ base.grid[0][nib] = c
+ }
+}
+
// processMounted folds each touched root-child subtree concurrently, stitches the resulting cells back into the base row, and folds the base up to the root.
func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Updates) ([]byte, error) {
pu := updates.parallel
@@ -149,7 +168,11 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up
path = append(path, byte(ni))
path = append(path, ch.ext...)
buildErr := dfsSubtreeDeep(w, ch, path, func(n *prefixNode, pth []byte, accountFresh bool) (cell, error) {
- return foldStorageRoot(gctx, foldSem, p.newStorageWorker, pu, n, pth, accountFresh)
+ sr, err := foldStorageRoot(gctx, foldSem, p.newStorageWorker, pu, n, pth, accountFresh)
+ if err == nil {
+ p.deepLocalFolds.Add(1)
+ }
+ return sr, err
})
if buildErr != nil {
w.Release()
diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go
index 58fb0006ff5..0e87ebe26e4 100644
--- a/execution/commitment/parallel_patricia_hashed.go
+++ b/execution/commitment/parallel_patricia_hashed.go
@@ -38,12 +38,16 @@ type ParallelPatriciaHashed struct {
rootHash atomic.Pointer[[]byte]
+ deepLocalFolds atomic.Uint64
+
leaveDeferredForCaller bool
deferredForCaller []*DeferredBranchUpdate
-
- streaming *StreamingCommitter
}
+// DeepLocalFolds reports how many big-storage accounts the last Process deep-folded
+// concurrently rather than demoting to serial recursion.
+func (p *ParallelPatriciaHashed) DeepLocalFolds() uint64 { return p.deepLocalFolds.Load() }
+
// NewParallelPatriciaHashed constructs a fresh ParallelPatriciaHashed.
func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int16, cfg TrieConfig) *ParallelPatriciaHashed {
p := &ParallelPatriciaHashed{
@@ -62,17 +66,11 @@ func (p *ParallelPatriciaHashed) SetNumWorkers(n int) {
n = runtime.NumCPU()
}
p.numWorkers = n
- if p.streaming != nil {
- p.streaming.SetNumWorkers(n)
- }
}
// SetLeaveDeferredForCaller makes Process leave the deferred branch updates for the caller to flush instead of applying them inline.
func (p *ParallelPatriciaHashed) SetLeaveDeferredForCaller(leave bool) {
p.leaveDeferredForCaller = leave
- if p.streaming != nil {
- p.streaming.SetLeaveDeferredForCaller(leave)
- }
}
// HasPendingDeferredUpdates reports whether Process left deferred branch updates for the caller to flush.
@@ -105,9 +103,7 @@ func (p *ParallelPatriciaHashed) Reset() {
p.template.Reset()
}
p.rootHash.Store(nil)
- if p.streaming != nil {
- p.streaming.Reset()
- }
+ p.deepLocalFolds.Store(0)
}
// Release frees the template and worker pool; the instance must not be used afterwards. Repeat calls are no-ops.
@@ -117,10 +113,6 @@ func (p *ParallelPatriciaHashed) Release() {
p.template = nil
}
p.rootHash.Store(nil)
- if p.streaming != nil {
- p.streaming.Release()
- p.streaming = nil
- }
}
// ResetContext propagates a new PatriciaContext to the template; per-worker contexts come from trieCtxFactory.
@@ -133,21 +125,10 @@ func (p *ParallelPatriciaHashed) ResetContext(ctx PatriciaContext) {
// SetTrieContextFactory replaces the per-worker context factory.
func (p *ParallelPatriciaHashed) SetTrieContextFactory(f TrieContextFactory) {
p.trieCtxFactory = f
- if p.streaming != nil {
- p.streaming.SetTrieContextFactory(f)
- }
-}
-
-// SetStreamingCommitter switches Process to the streaming path; the same committer must also be wired to the Updates buffer.
-func (p *ParallelPatriciaHashed) SetStreamingCommitter(sc *StreamingCommitter) {
- p.streaming = sc
- if sc != nil && p.trieCtxFactory != nil {
- sc.SetTrieContextFactory(p.trieCtxFactory)
- }
}
// syncWriter serializes concurrent trace writes from the root template and the
-// streaming/mount fold workers onto one underlying io.Writer.
+// mount fold workers onto one underlying io.Writer.
type syncWriter struct {
mu sync.Mutex
w io.Writer
@@ -155,8 +136,8 @@ type syncWriter struct {
// NewSyncWriter wraps w so concurrent trace writes are serialized. It is
// idempotent (a *syncWriter is returned unchanged) and returns nil for nil,
-// so the same guarded writer can be shared across the template and streaming
-// committer without stacking mutexes.
+// so the same guarded writer can be shared across the template and its fold
+// workers without stacking mutexes.
func NewSyncWriter(w io.Writer) io.Writer {
if w == nil {
return nil
@@ -211,16 +192,13 @@ func tracePrefix(w io.Writer, prefix string) io.Writer {
}
// SetTraceWriter routes trace output to w (nil disables tracing). The writer is
-// wrapped once in a mutex-guarded syncWriter shared with the template and the
-// streaming committer, whose fold workers fan it out to their per-goroutine tries.
+// wrapped once in a mutex-guarded syncWriter shared with the template, whose
+// fold workers fan it out to their per-goroutine tries.
func (p *ParallelPatriciaHashed) SetTraceWriter(w io.Writer) {
tw := NewSyncWriter(w)
if p.template != nil {
p.template.SetTraceWriter(tw)
}
- if p.streaming != nil {
- p.streaming.SetTraceWriter(tw)
- }
}
func (p *ParallelPatriciaHashed) EnableCsvMetrics(filePathPrefix string) {
@@ -230,9 +208,6 @@ func (p *ParallelPatriciaHashed) EnableCsvMetrics(filePathPrefix string) {
}
func (p *ParallelPatriciaHashed) Variant() TrieVariant {
- if p.streaming != nil {
- return VariantStreamingHexPatricia
- }
return VariantParallelHexPatricia
}
@@ -250,26 +225,6 @@ func (p *ParallelPatriciaHashed) RootHash() ([]byte, error) {
return p.template.RootHash()
}
-// processStreaming delegates Process to the attached StreamingCommitter and republishes the root.
-func (p *ParallelPatriciaHashed) processStreaming(ctx context.Context) ([]byte, error) {
- // The template root is the restore target of SetState, so it seeds the committer's base;
- // PromoteRootInto below keeps the two in sync after every fold.
- p.streaming.SeedRootFrom(p.template)
- rh, err := p.streaming.Process(ctx)
- if err != nil {
- return nil, err
- }
- if p.leaveDeferredForCaller {
- p.deferredForCaller = p.streaming.TakeDeferredUpdates()
- }
- // Promote the root into the template so EncodeCurrentState serializes a root SetState can restore.
- p.streaming.PromoteRootInto(p.template)
- out := make([]byte, len(rh))
- copy(out, rh)
- p.rootHash.Store(&out)
- return out, nil
-}
-
// Process is the entry point for parallel commitment computation; it requires updates.mode == ModeParallel.
func (p *ParallelPatriciaHashed) Process(
ctx context.Context,
@@ -289,11 +244,12 @@ func (p *ParallelPatriciaHashed) Process(
}
p.rootHash.Store(nil)
+ p.deepLocalFolds.Store(0)
pu := updates.parallel
if pu.trie == nil || pu.trie.root == nil || pu.trie.root.subtreeCount == 0 {
// A consumed (or never-touched) collection must return the carried root; folding
- // an empty streaming base would publish the empty-trie root instead.
+ // an empty base would publish the empty-trie root instead.
rh, rerr := p.template.RootHash()
if rerr != nil {
return nil, rerr
@@ -301,14 +257,6 @@ func (p *ParallelPatriciaHashed) Process(
return rh, nil
}
- if p.streaming != nil {
- rh, sErr := p.processStreaming(ctx)
- if sErr == nil {
- updates.consumeParallel()
- }
- return rh, sErr
- }
-
rh, mErr := p.processMounted(ctx, updates)
if mErr != nil {
pu.deferredMu.Lock()
diff --git a/execution/commitment/parallel_patricia_hashed_test.go b/execution/commitment/parallel_patricia_hashed_test.go
index a260cada6b3..0ef8484e1d3 100644
--- a/execution/commitment/parallel_patricia_hashed_test.go
+++ b/execution/commitment/parallel_patricia_hashed_test.go
@@ -931,7 +931,7 @@ func TestParallelReuseAcrossResetParity(t *testing.T) {
k2, u2 := sparseBatch2(k1, 3, false)
seqRoot, _ := incrementalRoot(t, modeSeq, 0, k1, u1, k2, u2)
- for _, variant := range []TrieVariant{VariantParallelHexPatricia, VariantStreamingHexPatricia} {
+ for _, variant := range []TrieVariant{VariantParallelHexPatricia} {
root := reusedInstanceIncrementalRoot(t, variant, 8, k1, u1, k2, u2)
require.Equalf(t, seqRoot, root, "reused-instance %s incremental root vs sequential", variant)
}
diff --git a/execution/commitment/parallel_streaming_bench_test.go b/execution/commitment/parallel_streaming_bench_test.go
index 114d7e36151..fd9f8ccb398 100644
--- a/execution/commitment/parallel_streaming_bench_test.go
+++ b/execution/commitment/parallel_streaming_bench_test.go
@@ -23,9 +23,7 @@ import (
"math/rand"
"runtime"
"slices"
- "sync/atomic"
"testing"
- "time"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
@@ -293,92 +291,6 @@ func Benchmark_StorageConcurrency(b *testing.B) {
}
}
-// Keeps burnCPU's result observable so the compiler cannot elide the synthetic work.
-var benchCPUSink atomic.Uint64
-
-// Synthetic per-touch CPU cost standing in for block execution.
-func burnCPU(iters int) {
- var x uint64 = 1469598103934665603
- for i := range iters {
- x = (x ^ uint64(i)) * 1099511628211
- }
- benchCPUSink.Add(x)
-}
-
-func streamingBenchCorpora() []struct {
- name string
- pk [][]byte
- upds []Update
-} {
- wk, wu := buildWhaleCorpus(bigAccountWhale(40_000))
- mk, mu := buildMixedCorpus(99, 20_000)
- return []struct {
- name string
- pk [][]byte
- upds []Update
- }{
- {"whale", wk, wu},
- {"mixed", mk, mu},
- }
-}
-
-// scheduler=true overlaps folds with the per-touch CPU burn; false defers all folds to Process.
-func runStreamingOverlapBench(b *testing.B, pk [][]byte, upds []Update, cpuIters int, scheduler bool) {
- ctx := context.Background()
- b.ReportAllocs()
- var (
- totalProcess time.Duration
- totalRefold uint64
- iters int
- )
- for b.Loop() {
- b.StopTimer()
- ms := NewMockState(b)
- ms.SetConcurrentCommitment(true)
- require.NoError(b, ms.applyPlainUpdates(pk, upds))
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- sc.SetNumWorkers(runtime.NumCPU())
- if scheduler {
- require.NoError(b, sc.StartScheduler(ctx))
- }
- b.StartTimer()
-
- for _, k := range pk {
- burnCPU(cpuIters)
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
- }
- procStart := time.Now()
- _, err := sc.Process(ctx)
- procDur := time.Since(procStart)
-
- b.StopTimer()
- require.NoError(b, err)
- totalProcess += procDur
- totalRefold += sc.RefoldCount()
- iters++
- sc.Release()
- b.StartTimer()
- }
- if iters > 0 {
- b.ReportMetric(float64(totalProcess.Nanoseconds())/float64(iters), "process-ns/op")
- b.ReportMetric(float64(totalRefold)/float64(iters), "refolds/op")
- }
-}
-
-// Mechanism sanity-check with synthetic CPU cost; numbers are not a performance claim.
-func Benchmark_StreamingOverlap(b *testing.B) {
- for _, c := range streamingBenchCorpora() {
- for _, cpu := range []int{0, 500, 5000} {
- b.Run(fmt.Sprintf("%s/cpu=%d/overlap", c.name, cpu), func(b *testing.B) {
- runStreamingOverlapBench(b, c.pk, c.upds, cpu, true)
- })
- b.Run(fmt.Sprintf("%s/cpu=%d/batch", c.name, cpu), func(b *testing.B) {
- runStreamingOverlapBench(b, c.pk, c.upds, cpu, false)
- })
- }
- }
-}
-
func Benchmark_DeepStorageWhale(b *testing.B) {
for _, slots := range []int{750_000} {
addr, accHash, accNib, accUpd, pk, upds, groups := whaleByNibble(slots)
diff --git a/execution/commitment/parallel_testkit_test.go b/execution/commitment/parallel_testkit_test.go
index bacf7d5ec77..d3614741c83 100644
--- a/execution/commitment/parallel_testkit_test.go
+++ b/execution/commitment/parallel_testkit_test.go
@@ -105,9 +105,6 @@ type runMode int
const (
modeSeq runMode = iota
modeParallel
- modeStreaming
- modeStreamingScheduled
- modeStreamingPublic
)
func newSeqTrie(t *testing.T, ms *MockState) *HexPatriciaHashed {
@@ -123,34 +120,6 @@ func newParTrie(t *testing.T, ms *MockState, workers int) *ParallelPatriciaHashe
return tr
}
-// Requires ms.SetConcurrentCommitment(true) already set.
-func newStreamCommitter(t *testing.T, ms *MockState, workers int, scheduler bool) *StreamingCommitter {
- t.Helper()
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- sc.SetNumWorkers(workers)
- if scheduler {
- require.NoError(t, sc.StartScheduler(context.Background()))
- }
- return sc
-}
-
-// newStreamingFixture builds a concurrent MockState with keys/upds applied and a StreamingCommitter
-// wired to it. Pass scheduler=true to start the background scheduler before returning.
-func newStreamingFixture(t *testing.T, keys [][]byte, upds []Update, workers int, scheduler ...bool) (*StreamingCommitter, *MockState) {
- t.Helper()
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- require.NoError(t, ms.applyPlainUpdates(keys, upds))
- sc := newStreamCommitter(t, ms, workers, len(scheduler) > 0 && scheduler[0])
- return sc, ms
-}
-
-func touchAll(sc *StreamingCommitter, keys [][]byte) {
- for _, k := range keys {
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
- }
-}
-
func processRoot(t *testing.T, trie Trie, ut *Updates) []byte {
t.Helper()
root, err := trie.Process(context.Background(), ut, "", nil, WarmupConfig{})
@@ -171,7 +140,6 @@ func processModeBatch(t *testing.T, ms *MockState, mode runMode, workers int, ke
// only through this blob.
func processModeBatchState(t *testing.T, ms *MockState, mode runMode, workers int, keys [][]byte, upds []Update, blob []byte) ([]byte, []byte) {
t.Helper()
- ctx := context.Background()
require.NoError(t, ms.applyPlainUpdates(keys, upds))
encoded := func(tr *HexPatriciaHashed) []byte {
@@ -187,48 +155,11 @@ func processModeBatchState(t *testing.T, ms *MockState, mode runMode, workers in
require.NoError(t, tr.RootTrie().SetState(blob))
ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
defer ut.Close()
- for i, k := range keys {
- ks := string(k)
- ut.TouchPlainKey(ks, nil, func(c *KeyUpdate, _ []byte) {
- c.plainKey = ks
- c.hashedKey = KeyToHexNibbleHash(k)
- c.update = &upds[i]
- })
- }
- return processRoot(t, tr, ut), encoded(tr.RootTrie())
- case modeStreaming, modeStreamingScheduled:
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- defer sc.Release()
- sc.SetNumWorkers(workers)
- tmpl := NewHexPatriciaHashed(length.Addr, ms, DefaultTrieConfig())
- defer tmpl.Release()
- require.NoError(t, tmpl.SetState(blob))
- sc.SeedRootFrom(tmpl)
- if mode == modeStreamingScheduled {
- require.NoError(t, sc.StartScheduler(context.Background()))
- }
+ // ModeParallel ignores the callback; values reach the trie via MockState.
for _, k := range keys {
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
+ ut.TouchPlainKey(string(k), nil, nil)
}
- r, err := sc.Process(ctx)
- require.NoError(t, err)
- sc.PromoteRootInto(tmpl)
- return bytes.Clone(r), encoded(tmpl)
- case modeStreamingPublic:
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut.Close()
- defer trie.Release()
- pt := trie.(*ParallelPatriciaHashed)
- pt.SetNumWorkers(workers)
- pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt.ResetContext(ms)
- require.NoError(t, pt.RootTrie().SetState(blob))
- for _, key := range keys {
- ut.TouchPlainKey(string(key), nil, ut.TouchAccount)
- }
- return processRoot(t, trie, ut), encoded(pt.RootTrie())
+ return processRoot(t, tr, ut), encoded(tr.RootTrie())
default:
tr := newSeqTrie(t, ms)
defer tr.Release()
@@ -239,6 +170,28 @@ func processModeBatchState(t *testing.T, ms *MockState, mode runMode, workers in
}
}
+// parallelBatchDeepFolds folds one batch through the parallel engine, additionally
+// reporting how many big-storage accounts took the concurrent deep fold rather than
+// demoting to serial recursion.
+func parallelBatchDeepFolds(t *testing.T, ms *MockState, workers int, keys [][]byte, upds []Update, blob []byte) ([]byte, []byte, uint64) {
+ t.Helper()
+ require.NoError(t, ms.applyPlainUpdates(keys, upds))
+
+ tr := newParTrie(t, ms, workers)
+ defer tr.Release()
+ require.NoError(t, tr.RootTrie().SetState(blob))
+ ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
+ defer ut.Close()
+ // ModeParallel ignores the callback; values reach the trie via MockState.
+ for _, k := range keys {
+ ut.TouchPlainKey(string(k), nil, nil)
+ }
+ root := processRoot(t, tr, ut)
+ encoded, err := tr.RootTrie().EncodeCurrentState(nil)
+ require.NoError(t, err)
+ return root, encoded, tr.DeepLocalFolds()
+}
+
func engineRoot(t *testing.T, mode runMode, workers int, keys [][]byte, upds []Update) ([]byte, *MockState) {
t.Helper()
ms := NewMockState(t)
@@ -248,6 +201,11 @@ func engineRoot(t *testing.T, mode runMode, workers int, keys [][]byte, upds []U
return processModeBatch(t, ms, mode, workers, keys, upds), ms
}
+func sequentialRoot(t *testing.T, keys [][]byte, upds []Update) ([]byte, *MockState) {
+ t.Helper()
+ return engineRoot(t, modeSeq, 0, keys, upds)
+}
+
// Folds two batches into one MockState so batch-1 branches become on-disk state for
// batch-2, with the trie state blob carried across the batches (encode/restore cycle).
func incrementalRoot(t *testing.T, mode runMode, workers int, k1 [][]byte, u1 []Update, k2 [][]byte, u2 []Update) ([]byte, *MockState) {
@@ -279,18 +237,6 @@ func requireAllEnginesParity(t *testing.T, k1 [][]byte, u1 []Update, k2 [][]byte
branchDiff(t, seqMs, parMs)
}
require.Equalf(t, seqRoot, parRoot, "parallel(workers=%d) vs sequential root mismatch", workers)
-
- strRoot, strMs := incrementalRoot(t, modeStreaming, workers, k1, u1, k2, u2)
- if !bytes.Equal(seqRoot, strRoot) {
- branchDiff(t, seqMs, strMs)
- }
- require.Equalf(t, seqRoot, strRoot, "streaming(workers=%d) vs sequential root mismatch", workers)
-
- schRoot, schMs := incrementalRoot(t, modeStreamingScheduled, workers, k1, u1, k2, u2)
- if !bytes.Equal(seqRoot, schRoot) {
- branchDiff(t, seqMs, schMs)
- }
- require.Equalf(t, seqRoot, schRoot, "streaming-scheduled(workers=%d) vs sequential root mismatch", workers)
}
func requireBranchParity(t *testing.T, seq, got *MockState) {
diff --git a/execution/commitment/parallel_trace_test.go b/execution/commitment/parallel_trace_test.go
index 8df40baf4fe..1d3070f82dc 100644
--- a/execution/commitment/parallel_trace_test.go
+++ b/execution/commitment/parallel_trace_test.go
@@ -44,7 +44,7 @@ func TestTracePrefix_NilDisables(t *testing.T) {
require.Nil(t, tracePrefix(nil, "[x] "))
}
-// Concurrent fold workers (the parallel/streaming trie) each own a prefixWriter
+// Concurrent fold workers (the parallel trie) each own a prefixWriter
// over one shared syncWriter; every emitted line must stay whole and carry its
// worker tag — never interleaved or corrupted mid-line.
func TestSyncWriter_ConcurrentLinesStayAttributed(t *testing.T) {
diff --git a/execution/commitment/state_roundtrip_regression_test.go b/execution/commitment/state_roundtrip_regression_test.go
index e53698f984c..7519ad7007e 100644
--- a/execution/commitment/state_roundtrip_regression_test.go
+++ b/execution/commitment/state_roundtrip_regression_test.go
@@ -17,7 +17,6 @@
package commitment
import (
- "context"
"encoding/hex"
"testing"
@@ -43,9 +42,6 @@ func requireRestartParity(t *testing.T, batches []engineBatch, combinedK [][]byt
}{
{"parallel_w1", modeParallel, 1},
{"parallel_w4", modeParallel, 4},
- {"streaming_committer_w4", modeStreaming, 4},
- {"streaming_scheduled_w4", modeStreamingScheduled, 4},
- {"streaming_public_w4", modeStreamingPublic, 4},
} {
roots, _ := runEngineBatches(t, tc.mode, tc.workers, batches)
for i := range batches {
@@ -242,95 +238,6 @@ func TestSetState_RepairsLegacyRootPresent(t *testing.T) {
require.Equal(t, oracle, got, "legacy rootPresent=false blob dropped the carried state")
}
-// Restoring the template AFTER a scheduler already built its base must not fold against
-// the stale base: the changed seed drops it so Process rebuilds from the restored root.
-func TestStateRoundTrip_SeedAfterSchedulerStart(t *testing.T) {
- t.Parallel()
- k1, u1, k2, u2, kc, uc := singleNibbleCorpus()
- oracle, _ := engineRoot(t, modeSeq, 0, kc, uc)
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- _, blob := processModeBatchState(t, ms, modeStreaming, 4, k1, u1, nil)
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- defer sc.Release()
- sc.SetNumWorkers(4)
- require.NoError(t, sc.StartScheduler(context.Background()))
-
- tmpl := NewHexPatriciaHashed(length.Addr, ms, DefaultTrieConfig())
- defer tmpl.Release()
- require.NoError(t, tmpl.SetState(blob))
- sc.SeedRootFrom(tmpl)
-
- for _, k := range k2 {
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
- }
- got, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, oracle, got, "seed arriving after StartScheduler was folded against the stale base")
-}
-
-// A background fold completed against the previous seed's base must not be stitched after
-// the seed changes: SeedRootFrom invalidates folded splits so Process re-folds them.
-func TestStateRoundTrip_SeedChangeInvalidatesFoldedSplits(t *testing.T) {
- t.Parallel()
- var addrs []string
- for i := range 6 {
- addrs = append(addrs, addrHex(findAddressForNibble(7, 500+i)))
- }
- ub1 := NewUpdateBuilder()
- for i, a := range addrs {
- ub1.Balance(a, uint64(100+i))
- }
- k1, u1 := ub1.Build()
-
- ub2 := NewUpdateBuilder().Balance(addrs[0], 9100).Balance(addrs[1], 9200).
- Balance(addrHex(findAddressForNibble(7, 5100)), 51).Balance(addrHex(findAddressForNibble(7, 5200)), 52)
- k2, u2 := ub2.Build()
-
- ubc := NewUpdateBuilder().Balance(addrs[0], 9100).Balance(addrs[1], 9200)
- for i, a := range addrs[2:] {
- ubc.Balance(a, uint64(102+i))
- }
- ubc.Balance(addrHex(findAddressForNibble(7, 5100)), 51).Balance(addrHex(findAddressForNibble(7, 5200)), 52)
- kc, uc := ubc.Build()
- oracle, _ := engineRoot(t, modeSeq, 0, kc, uc)
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- _, blob := processModeBatchState(t, ms, modeStreaming, 4, k1, u1, nil)
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- sc := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- defer sc.Release()
- sc.SetNumWorkers(4)
- require.NoError(t, sc.StartScheduler(context.Background()))
- for _, k := range k2 {
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
- }
- // fold the touched split against the scheduler's unseeded base, synchronously, leaving
- // it in the reusable folded state a background fold would produce
- sc.foldSplitBg(7)
- sc.trieMu.RLock()
- s := sc.splits[byte(7)]
- sc.trieMu.RUnlock()
- require.NotNil(t, s)
- s.mu.Lock()
- require.True(t, s.reusable(), "precondition: split folded against the stale base")
- s.mu.Unlock()
-
- tmpl := NewHexPatriciaHashed(length.Addr, ms, DefaultTrieConfig())
- defer tmpl.Release()
- require.NoError(t, tmpl.SetState(blob))
- sc.SeedRootFrom(tmpl)
-
- got, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, oracle, got, "a split folded against the stale base was stitched after the seed changed")
-}
-
// A fresh trie with NO carried state blob must still bootstrap from the on-disk branch
// records alone when the root is a real branch — the rebuild/reset shape the blob-carrying
// helpers no longer exercise.
@@ -368,8 +275,6 @@ func TestFreshTrieBootstrapOverDiskState(t *testing.T) {
workers int
}{
{"parallel_w4", modeParallel, 4},
- {"streaming_committer_w4", modeStreaming, 4},
- {"streaming_public_w4", modeStreamingPublic, 4},
} {
require.Equalf(t, want, bootstrapRoot(tc.mode, tc.workers), "%s: blobless bootstrap diverged", tc.name)
}
diff --git a/execution/commitment/streaming_commitment.go b/execution/commitment/streaming_commitment.go
deleted file mode 100644
index 6c530a46753..00000000000
--- a/execution/commitment/streaming_commitment.go
+++ /dev/null
@@ -1,962 +0,0 @@
-// Copyright 2026 The Erigon Authors
-// This file is part of Erigon.
-//
-// Erigon is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// Erigon is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with Erigon. If not, see .
-
-package commitment
-
-import (
- "bytes"
- "context"
- "errors"
- "fmt"
- "io"
- "math/bits"
- "runtime"
- "sync"
- "sync/atomic"
-
- "golang.org/x/sync/errgroup"
- "golang.org/x/sync/semaphore"
-
- "github.com/erigontech/erigon/db/kv"
-)
-
-type splitState struct {
- prefix []byte
- cell cell
- deferred []*DeferredBranchUpdate
- gen uint64
- keyCount uint64
- lastFoldedSize uint64
- dirty bool
- folded bool
- queued bool
- mu sync.Mutex
-}
-
-// reusable reports a cell cached by a background fold and not yet invalidated by
-// a later touch. Callers must hold s.mu.
-func (s *splitState) reusable() bool { return s.folded && !s.dirty }
-
-const defaultEagerFold = 256
-
-// shouldEagerFold re-folds only once the split's key count has at least doubled
-// since its last fold (and cleared the floor), keeping total re-fold work linear.
-// Callers hold s.mu.
-func (sc *StreamingCommitter) shouldEagerFold(s *splitState) bool {
- return s.keyCount >= sc.eagerFloor && s.keyCount >= 2*s.lastFoldedSize
-}
-
-// SetEagerFold overrides the coalescing floor (default defaultEagerFold).
-func (sc *StreamingCommitter) SetEagerFold(n uint64) { sc.eagerFloor = n }
-
-// StreamingCommitter overlaps commitment fold work with block execution.
-type StreamingCommitter struct {
- trieCtxFactory TrieContextFactory
- cfg TrieConfig
- accountKeyLen int16
- numWorkers int
-
- trie *prefixTrie
- splits map[byte]*splitState
- eagerFloor uint64
-
- // trieMu serializes prefix-trie mutation against the scheduler's structural reads.
- trieMu sync.RWMutex
-
- started atomic.Bool
- quit chan struct{}
- wg sync.WaitGroup
- bgCtx context.Context
- dirtyCh chan byte
- base *HexPatriciaHashed
- baseCleanup func()
- refoldTotal atomic.Uint64
- inFlight atomic.Int64
-
- // foldGate, when set, is a test seam invoked just before a background fold.
- foldGate func(nib byte)
-
- leaveDeferredForCaller bool
- deferredForCaller []*DeferredBranchUpdate
-
- deepLocalFolds atomic.Uint64
-
- // rootValid gates root promotion: cleared each Process and set only on the
- // folded path, so the no-touch path leaves the template's prior root untouched.
- rootCell cell
- rootChecked bool
- rootTouched bool
- rootPresent bool
- rootValid bool
- // rootSeeded marks the same snapshot as a base seed: a trie whose root row
- // collapsed to one child has no root branch record on disk, so the carried
- // root cell is the only way a fresh base can see the existing state.
- rootSeeded bool
-
- traceW io.Writer
-}
-
-func (sc *StreamingCommitter) SetTraceWriter(w io.Writer) { sc.traceW = NewSyncWriter(w) }
-
-// NewStreamingCommitter constructs a StreamingCommitter ready to accept touches.
-func NewStreamingCommitter(ctxFactory TrieContextFactory, accountKeyLen int16, cfg TrieConfig) *StreamingCommitter {
- sc := &StreamingCommitter{
- trieCtxFactory: ctxFactory,
- cfg: cfg,
- accountKeyLen: accountKeyLen,
- numWorkers: runtime.NumCPU(),
- trie: newPrefixTrie(),
- splits: make(map[byte]*splitState),
- eagerFloor: defaultEagerFold,
- }
- return sc
-}
-
-// SetNumWorkers overrides the worker count for the next Process call. Values
-// <= 0 fall back to runtime.NumCPU.
-func (sc *StreamingCommitter) SetNumWorkers(n int) {
- if n <= 0 {
- n = runtime.NumCPU()
- }
- sc.numWorkers = n
-}
-
-// SetTrieContextFactory replaces the per-worker context factory.
-func (sc *StreamingCommitter) SetTrieContextFactory(f TrieContextFactory) {
- sc.trieCtxFactory = f
-}
-
-// SetLeaveDeferredForCaller makes Process leave the accumulated deferred branch
-// updates for the caller to flush instead of applying them inline.
-func (sc *StreamingCommitter) SetLeaveDeferredForCaller(leave bool) {
- sc.leaveDeferredForCaller = leave
-}
-
-// TakeDeferredUpdates returns the deferred branch updates staged for the caller
-// and clears them; the caller takes ownership and returns them to the pool.
-func (sc *StreamingCommitter) TakeDeferredUpdates() []*DeferredBranchUpdate {
- d := sc.deferredForCaller
- sc.deferredForCaller = nil
- return d
-}
-
-// TouchKey records a touched key. hashedKey is in nibble form; plainKey/update
-// backing must stay stable until Process, and a nil update makes the fold
-// re-read the value from ctx.
-func (sc *StreamingCommitter) TouchKey(hashedKey, plainKey []byte, update *Update) {
- sc.trieMu.Lock()
- isNew := sc.trie.Insert(hashedKey, plainKey, update)
- if len(hashedKey) == 0 {
- sc.trieMu.Unlock()
- return
- }
- nib := hashedKey[0]
- s := sc.splits[nib]
- if s == nil {
- s = &splitState{prefix: []byte{nib}}
- sc.splits[nib] = s
- }
- s.mu.Lock()
- s.dirty = true
- s.gen++
- if isNew {
- s.keyCount++
- }
- enqueue := sc.started.Load() && !s.queued && sc.shouldEagerFold(s)
- if enqueue {
- s.queued = true
- }
- s.mu.Unlock()
- sc.trieMu.Unlock()
-
- if enqueue {
- sc.enqueue(nib)
- }
-}
-
-// Process folds every touched top-nibble split into a cell, stitches the cells
-// into the base row, and folds to the root.
-func (sc *StreamingCommitter) Process(ctx context.Context) ([]byte, error) {
- if sc.trieCtxFactory == nil {
- return nil, errors.New("StreamingCommitter.Process requires a TrieContextFactory")
- }
- if sc.trie == nil {
- return nil, errors.New("StreamingCommitter.Process called after Release")
- }
-
- sc.Stop()
- sc.rootValid = false
-
- base, cleanup, root, err := sc.processBase(ctx)
- if err != nil {
- return nil, err
- }
- defer cleanup()
-
- if root == nil || root.subtreeCount == 0 {
- return base.RootHash()
- }
-
- present, err := sc.foldPresentSplits(ctx, base, root)
- if err != nil {
- sc.dropSplitDeferred()
- return nil, err
- }
-
- var (
- cells [16]cell
- deferred []*DeferredBranchUpdate
- )
- for nib := range 16 {
- if !present[nib] {
- continue
- }
- s := sc.splits[byte(nib)]
- cells[nib] = s.cell
- deferred = append(deferred, s.deferred...)
- s.deferred = nil
- }
-
- stitchSplitCells(base, &cells, &present)
-
- if base.activeRows == 0 {
- base.activeRows = 1
- }
- for base.activeRows > 0 {
- if err := ctx.Err(); err != nil {
- return nil, err
- }
- if err := base.fold(); err != nil {
- return nil, fmt.Errorf("StreamingCommitter: root fold: %w", err)
- }
- }
- if d := base.TakeDeferredUpdates(); len(d) > 0 {
- deferred = mergeDeferredByPrefix(deferred, d)
- }
-
- if sc.leaveDeferredForCaller {
- sc.deferredForCaller = deferred
- } else if err := sc.applyDeferred(ctx, deferred); err != nil {
- return nil, err
- }
- sc.captureRoot(base)
- rh, err := base.RootHash()
- if err != nil {
- return nil, err
- }
- flushTrieStateRates()
- sc.endBlock()
- return rh, nil
-}
-
-// endBlock drains the per-block touch funnel and releases the scheduler base
-// (Process folds it down to the terminal root, so it cannot be reused), keeping
-// the worker pool and the caller's staged root/deferred snapshots.
-func (sc *StreamingCommitter) endBlock() {
- if sc.trie != nil {
- sc.trie.Reset()
- }
- sc.dropSplitDeferred()
- clear(sc.splits)
- sc.releaseBase()
-}
-
-// captureRoot snapshots the base trie's terminal root cell and flags by value
-// so the snapshot survives the base being released.
-func (sc *StreamingCommitter) captureRoot(base *HexPatriciaHashed) {
- sc.rootCell = base.root
- sc.rootChecked = base.rootChecked
- sc.rootTouched = base.rootTouched
- sc.rootPresent = base.rootPresent
- sc.rootValid = true
- sc.rootSeeded = true
-}
-
-// SeedRootFrom snapshots tmpl's root cell and flags as the seed for the next
-// Process's base — the mirror of PromoteRootInto, used after tmpl was restored
-// via SetState. A changed seed invalidates any base built from the previous one
-// (including a running scheduler's), which is dropped so the next Process
-// rebuilds from the new seed instead of folding against stale root state.
-func (sc *StreamingCommitter) SeedRootFrom(tmpl *HexPatriciaHashed) {
- if tmpl == nil {
- return
- }
- if sc.rootCell == tmpl.root && sc.rootChecked == tmpl.rootChecked &&
- sc.rootTouched == tmpl.rootTouched && sc.rootPresent == tmpl.rootPresent {
- sc.rootSeeded = true
- return
- }
- sc.rootCell = tmpl.root
- sc.rootChecked = tmpl.rootChecked
- sc.rootTouched = tmpl.rootTouched
- sc.rootPresent = tmpl.rootPresent
- sc.rootSeeded = true
- if sc.base != nil {
- sc.Stop()
- sc.releaseBase()
- }
- // Splits folded against the previous seed's base are stale; drop their cells and
- // deferred updates so Process re-folds them against the reseeded base. Read-lock
- // trieMu: TouchKey inserts into sc.splits under its write lock, and iterating a
- // map against a concurrent insert is a fatal runtime error.
- sc.trieMu.RLock()
- for _, s := range sc.splits {
- s.mu.Lock()
- s.folded = false
- s.dirty = true
- for _, upd := range s.deferred {
- putDeferredUpdate(upd)
- }
- s.deferred = nil
- s.mu.Unlock()
- }
- sc.trieMu.RUnlock()
-}
-
-// PromoteRootInto copies the most recently folded root cell and flags into tmpl,
-// reporting whether a fold result was promoted; the no-touch path returns false
-// and leaves the template's prior root in place.
-func (sc *StreamingCommitter) PromoteRootInto(tmpl *HexPatriciaHashed) bool {
- if !sc.rootValid || tmpl == nil {
- return false
- }
- tmpl.root = sc.rootCell
- tmpl.rootChecked = sc.rootChecked
- tmpl.rootTouched = sc.rootTouched
- tmpl.rootPresent = sc.rootPresent
- return true
-}
-
-// newProcessBase builds the per-Process base trie, unfolded at the root unless
-// the prefix trie is empty (no touches).
-func (sc *StreamingCommitter) newProcessBase(ctx context.Context) (*HexPatriciaHashed, func(), *prefixNode, error) {
- root := sc.trie.root
- if root == nil || root.subtreeCount == 0 {
- base, cleanup := sc.newBaseTrie(ctx)
- return base, cleanup, root, nil
- }
- if len(root.ext) != 0 {
- return nil, nil, nil, fmt.Errorf("StreamingCommitter: root.ext len %d not yet supported", len(root.ext))
- }
- base, cleanup, err := sc.buildBase(ctx)
- if err != nil {
- return nil, nil, nil, err
- }
- return base, cleanup, root, nil
-}
-
-// newBaseTrie constructs a fresh deferring base trie, seeded with the carried
-// root snapshot when one exists, and a cleanup releasing it.
-func (sc *StreamingCommitter) newBaseTrie(ctx context.Context) (*HexPatriciaHashed, func()) {
- base := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg)
- bctx, bclean := sc.trieCtxFactory(ctx)
- base.ResetContext(bctx)
- base.SetTraceWriter(sc.traceW)
- base.branchEncoder.setDeferUpdates(true)
- base.SetLeaveDeferredForCaller(true)
- if sc.rootSeeded {
- base.root = sc.rootCell
- base.rootChecked = sc.rootChecked
- base.rootTouched = sc.rootTouched
- base.rootPresent = sc.rootPresent
- }
- return base, func() {
- base.Release()
- if bclean != nil {
- bclean()
- }
- }
-}
-
-// processBase returns the base trie Process folds and stitches into, reusing the
-// persistent scheduler base when one exists (its cleanup is then a no-op).
-func (sc *StreamingCommitter) processBase(ctx context.Context) (*HexPatriciaHashed, func(), *prefixNode, error) {
- if sc.base != nil {
- root := sc.trie.root
- if root != nil && len(root.ext) != 0 {
- return nil, nil, nil, fmt.Errorf("StreamingCommitter: root.ext len %d not yet supported", len(root.ext))
- }
- return sc.base, func() {}, root, nil
- }
- return sc.newProcessBase(ctx)
-}
-
-// buildBase builds a base trie unfolded one level at the on-disk root so its
-// row 0 carries every top-nibble sibling the split cells stitch into.
-func (sc *StreamingCommitter) buildBase(ctx context.Context) (*HexPatriciaHashed, func(), error) {
- base, cleanup := sc.newBaseTrie(ctx)
-
- if err := unfoldRootWall(ctx, base); err != nil {
- cleanup()
- return nil, nil, fmt.Errorf("StreamingCommitter: unfold root: %w", err)
- }
- seedRootBase(base)
- return base, cleanup, nil
-}
-
-// SetFoldGate installs a test seam invoked just before a background fold.
-func (sc *StreamingCommitter) SetFoldGate(fn func(nib byte)) { sc.foldGate = fn }
-
-// RefoldCount reports how many background folds were discarded as wasted work.
-func (sc *StreamingCommitter) RefoldCount() uint64 { return sc.refoldTotal.Load() }
-
-// StartScheduler builds the persistent base and launches the background fold
-// pool; after it returns TouchKey enqueues dirtied splits. Calling it twice is a no-op.
-func (sc *StreamingCommitter) StartScheduler(ctx context.Context) error {
- if sc.trieCtxFactory == nil {
- return errors.New("StreamingCommitter.StartScheduler requires a TrieContextFactory")
- }
- if sc.started.Load() {
- return nil
- }
- sc.releaseBase()
- base, cleanup, err := sc.buildBase(ctx)
- if err != nil {
- return err
- }
- sc.base = base
- sc.baseCleanup = cleanup
- sc.bgCtx = ctx
- sc.quit = make(chan struct{})
- sc.dirtyCh = make(chan byte, 256)
- sc.started.Store(true)
-
- for range sc.numWorkers {
- sc.wg.Go(sc.scheduleWorker)
- }
- return nil
-}
-
-// Stop drains the background fold pool, waiting for any in-flight fold to finish.
-// Safe to call when no scheduler is running and to call twice.
-func (sc *StreamingCommitter) Stop() {
- if !sc.started.CompareAndSwap(true, false) {
- return
- }
- close(sc.quit)
- sc.wg.Wait()
-}
-
-func (sc *StreamingCommitter) scheduleWorker() {
- for {
- select {
- case <-sc.quit:
- return
- case nib := <-sc.dirtyCh:
- sc.inFlight.Add(1)
- sc.foldSplitBg(nib)
- sc.inFlight.Add(-1)
- }
- }
-}
-
-// enqueue offers a dirtied split to the fold pool without blocking; a full queue
-// just leaves the split dirty for Process, losing overlap but not safety.
-func (sc *StreamingCommitter) enqueue(nib byte) {
- if !sc.started.Load() {
- return
- }
- select {
- case sc.dirtyCh <- nib:
- default:
- }
-}
-
-// touchedKey is a snapshotted touch a background fold replays; hk is copied off
-// the walk path while pk/upd reference the caller's stable backing.
-type touchedKey struct {
- hk []byte
- pk []byte
- upd *Update
-}
-
-// foldSplitBg folds one split against an isolating overlay, installing the result
-// only if no touch bumped the split's gen and it did not self-flush meanwhile.
-func (sc *StreamingCommitter) foldSplitBg(nib byte) {
- sc.trieMu.RLock()
- root := sc.trie.root
- child, ok := childForNib(root, nib)
- s := sc.splits[nib]
- if !ok || s == nil {
- if s != nil {
- s.mu.Lock()
- s.queued = false
- s.mu.Unlock()
- }
- sc.trieMu.RUnlock()
- return
- }
-
- keys := collectSplitKeys(child, nib)
- s.mu.Lock()
- genStart := s.gen
- // Close the coalescing gate at fold start (snapshot size), not end, or a stale
- // lastFoldedSize would let every mid-fold touch re-enqueue for the fold's duration.
- s.lastFoldedSize = uint64(len(keys))
- s.queued = false
- s.mu.Unlock()
- sc.trieMu.RUnlock()
-
- if sc.foldGate != nil {
- sc.foldGate(nib)
- }
-
- c, deferred, flushed, err := sc.foldKeys(nib, keys)
-
- s.mu.Lock()
- if err != nil || flushed || s.gen != genStart {
- for _, upd := range deferred {
- putDeferredUpdate(upd)
- }
- sc.refoldTotal.Add(1)
- // Re-fold a mid-fold-touched split only if the gate still passes, bounding
- // a streaming whale to O(N) instead of O(N^2) re-folds.
- reEnqueue := s.gen != genStart && sc.shouldEagerFold(s)
- s.mu.Unlock()
- if reEnqueue {
- sc.markQueued(s, nib)
- }
- return
- }
- for _, upd := range s.deferred {
- putDeferredUpdate(upd)
- }
- s.deferred = deferred
- s.cell = c
- s.folded = true
- s.dirty = false
- s.mu.Unlock()
-}
-
-// markQueued re-enqueues a split, deduped so a burst of touches schedules it once.
-func (sc *StreamingCommitter) markQueued(s *splitState, nib byte) {
- s.mu.Lock()
- if s.queued {
- s.mu.Unlock()
- return
- }
- s.queued = true
- s.mu.Unlock()
- sc.enqueue(nib)
-}
-
-// foldKeys folds a snapshotted split's keys on a pooled worker whose overlay ctx
-// discards branch writes; the returned flushed flag reports a mid-fold self-flush.
-func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*DeferredBranchUpdate, bool, error) {
- w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg)
- w.mountTo(sc.base, int(nib))
- if sc.traceW != nil {
- w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[fold %x] ", nib)))
- } else {
- w.SetTraceWriter(nil)
- }
- rctx, cleanup := sc.trieCtxFactory(sc.bgCtx)
- if cleanup != nil {
- defer cleanup()
- }
- ov := &overlayContext{base: rctx}
- w.ResetContext(ov)
- w.branchEncoder.setDeferUpdates(true)
- w.SetLeaveDeferredForCaller(true)
-
- var err error
- for i := range keys {
- if err = w.followAndUpdate(keys[i].hk, keys[i].pk, keys[i].upd); err != nil {
- break
- }
- }
- var c cell
- if err == nil {
- c, err = w.foldMounted(sc.bgCtx, int(nib))
- }
- deferred := w.TakeDeferredUpdates()
- w.Release()
- return c, deferred, ov.flushed, err
-}
-
-// childForNib returns the top-nibble split-point child of root, or false if the
-// nibble carries no touched keys.
-func childForNib(root *prefixNode, nib byte) (*prefixNode, bool) {
- if root == nil || len(root.ext) != 0 {
- return nil, false
- }
- idx, ok := childIndex(root, nib)
- if !ok {
- return nil, false
- }
- return root.children[idx], true
-}
-
-// keyArena copies walk-path nibbles into chunked backing buffers so each
-// collected key gets a stable slice without one allocation per key. remaining
-// is the caller's expected key count; without it a subtree far smaller than a
-// chunk still burns a whole chunk, and most subtrees are.
-type keyArena struct {
- buf []byte
- remaining int
-}
-
-const keyArenaChunk = 64 * 1024
-
-func (a *keyArena) copy(hk []byte) []byte {
- if len(hk) > cap(a.buf)-len(a.buf) {
- want := len(hk) * max(a.remaining, 1)
- a.buf = make([]byte, 0, max(min(want, keyArenaChunk), len(hk)))
- }
- a.remaining--
- start := len(a.buf)
- a.buf = append(a.buf, hk...)
- return a.buf[start:len(a.buf):len(a.buf)]
-}
-
-// collectSplitKeys walks a split's subtree in sorted order, copying each key's
-// hashed nibbles off the reused walk path.
-func collectSplitKeys(child *prefixNode, nib byte) []touchedKey {
- path := make([]byte, 0, 144)
- path = append(path, nib)
- path = append(path, child.ext...)
- return collectSubtreeKeys(child, path)
-}
-
-// overlayContext isolates a background fold: writes never reach the real store
-// but a self-flushed prefix re-reads its own write, and flushed records that.
-type overlayContext struct {
- base PatriciaContext
- writes map[string][]byte
- flushed bool
-}
-
-func (o *overlayContext) Branch(prefix []byte) ([]byte, kv.Step, error) {
- if o.writes != nil {
- if d, ok := o.writes[string(prefix)]; ok {
- return d, 0, nil
- }
- }
- return o.base.Branch(prefix)
-}
-
-func (o *overlayContext) PutBranch(prefix, data, _ []byte) error {
- if o.writes == nil {
- o.writes = make(map[string][]byte)
- }
- o.writes[string(prefix)] = bytes.Clone(data)
- o.flushed = true
- return nil
-}
-
-func (o *overlayContext) Account(plainKey []byte) (*Update, error) { return o.base.Account(plainKey) }
-func (o *overlayContext) Storage(plainKey []byte) (*Update, error) { return o.base.Storage(plainKey) }
-
-// foldPresentSplits re-folds every touched top-nibble split concurrently onto the
-// base, recording which slots were folded; it never applies or merges.
-func (sc *StreamingCommitter) foldPresentSplits(ctx context.Context, base *HexPatriciaHashed, root *prefixNode) ([16]bool, error) {
- var present [16]bool
- foldSem := newFoldSem()
- g, gctx := errgroup.WithContext(ctx)
- g.SetLimit(min(sc.numWorkers, maxFoldConcurrency()))
-
- childIdx := 0
- for bm := root.bitmap; bm != 0; {
- nib := bits.TrailingZeros16(bm)
- child := root.children[childIdx]
- ni := byte(nib)
- s := sc.splits[ni]
- if s == nil {
- s = &splitState{prefix: []byte{ni}}
- sc.splits[ni] = s
- }
- present[nib] = true
- s.mu.Lock()
- reuse := s.reusable()
- s.mu.Unlock()
- if reuse {
- childIdx++
- bm &^= uint16(1) << nib
- continue
- }
- ch := child
- g.Go(func() error { return sc.foldSplit(gctx, foldSem, base, s, ch) })
- childIdx++
- bm &^= uint16(1) << nib
- }
- if err := g.Wait(); err != nil {
- return present, err
- }
- return present, nil
-}
-
-// foldDirtySplits re-folds every touched split without merging to the root or
-// writing to the store. Repeated calls are re-fold-invariant only while no touched
-// branch collapses, since a collapse self-flushes mid-fold and a second fold
-// would double-apply.
-func (sc *StreamingCommitter) foldDirtySplits(ctx context.Context) error {
- if sc.trieCtxFactory == nil {
- return errors.New("StreamingCommitter.foldDirtySplits requires a TrieContextFactory")
- }
- base, cleanup, root, err := sc.newProcessBase(ctx)
- if err != nil {
- return err
- }
- defer cleanup()
- if root == nil || root.subtreeCount == 0 {
- return nil
- }
- _, err = sc.foldPresentSplits(ctx, base, root)
- return err
-}
-
-// stitchSplitCells drops each folded split cell into the base row at its top-nibble slot;
-// foldMounted already returns cells excluding the mount nibble, so they are stitched verbatim.
-func stitchSplitCells(base *HexPatriciaHashed, cells *[16]cell, present *[16]bool) {
- for nib := range 16 {
- if !present[nib] {
- continue
- }
- c := cells[nib]
- base.touchMap[0] |= uint16(1) << nib
- if !c.IsEmpty() {
- base.afterMap[0] |= uint16(1) << nib
- } else {
- base.afterMap[0] &^= uint16(1) << nib
- }
- base.depths[0] = 1
- base.grid[0][nib] = c
- }
-}
-
-// foldSplit re-folds one top-nibble subtree on a worker mounted at the unfolded
-// base, to the split cell rather than the root, replacing the split's cell and
-// deferred set.
-func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.Weighted, base *HexPatriciaHashed, s *splitState, child *prefixNode) error {
- ni := s.prefix[0]
- w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg)
- w.mountTo(base, int(ni))
- if sc.traceW != nil {
- w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[split %x] ", ni)))
- } else {
- w.SetTraceWriter(nil)
- }
- wctx, cleanup := sc.trieCtxFactory(ctx)
- if cleanup != nil {
- defer cleanup()
- }
- w.ResetContext(wctx)
- w.branchEncoder.setDeferUpdates(true)
- w.SetLeaveDeferredForCaller(true)
-
- var pu parallelUpdate
- path := make([]byte, 0, 144)
- path = append(path, ni)
- path = append(path, child.ext...)
- deepStorageRoot := func(n *prefixNode, pth []byte, accountFresh bool) (cell, error) {
- sr, err := foldStorageRoot(ctx, foldSem, sc.newStorageWorker, &pu, n, pth, accountFresh)
- if err == nil {
- sc.deepLocalFolds.Add(1)
- }
- return sr, err
- }
- if err := dfsSubtreeDeep(w, child, path, deepStorageRoot); err != nil {
- w.Release()
- for _, upd := range pu.deferredCombined {
- putDeferredUpdate(upd)
- }
- return fmt.Errorf("split[%x] build: %w", ni, err)
- }
- c, err := w.foldMounted(ctx, int(ni))
- if err != nil {
- w.Release()
- for _, upd := range pu.deferredCombined {
- putDeferredUpdate(upd)
- }
- return fmt.Errorf("split[%x] fold: %w", ni, err)
- }
-
- newDeferred := pu.deferredCombined
- if d := w.TakeDeferredUpdates(); len(d) > 0 {
- newDeferred = append(newDeferred, d...)
- }
- w.Release()
-
- s.mu.Lock()
- for _, upd := range s.deferred {
- putDeferredUpdate(upd)
- }
- s.deferred = newDeferred
- s.cell = c
- s.dirty = false
- s.mu.Unlock()
- return nil
-}
-
-// DeepLocalFolds reports how many big-storage accounts the streaming path deep-folded.
-func (sc *StreamingCommitter) DeepLocalFolds() uint64 { return sc.deepLocalFolds.Load() }
-
-// newStorageWorker sources a concurrent-storage-fold worker; disjoint subtree
-// prefixes keep a mid-fold self-flush from racing another fold's writes.
-func (sc *StreamingCommitter) newStorageWorker(ctx context.Context) (*HexPatriciaHashed, func()) {
- return newDeferredStorageWorker(ctx, sc.accountKeyLen, sc.cfg, sc.trieCtxFactory, sc.traceW)
-}
-
-// dropSplitDeferred returns every split's staged deferred branch updates to the pool.
-func (sc *StreamingCommitter) dropSplitDeferred() {
- for _, s := range sc.splits {
- for _, upd := range s.deferred {
- putDeferredUpdate(upd)
- }
- s.deferred = nil
- }
-}
-
-// mergeDeferredByPrefix combines two deferred-update slices, keeping newer's
-// entry for any prefix both supply and recycling the superseded older one.
-func mergeDeferredByPrefix(older, newer []*DeferredBranchUpdate) []*DeferredBranchUpdate {
- if len(older) == 0 {
- return newer
- }
- inNewer := make(map[string]struct{}, len(newer))
- for _, u := range newer {
- inNewer[string(u.prefix)] = struct{}{}
- }
- out := newer
- for _, u := range older {
- if _, ok := inNewer[string(u.prefix)]; ok {
- putDeferredUpdate(u)
- continue
- }
- out = append(out, u)
- }
- return out
-}
-
-func (sc *StreamingCommitter) applyDeferred(ctx context.Context, deferred []*DeferredBranchUpdate) error {
- defer func() {
- for _, upd := range deferred {
- putDeferredUpdate(upd)
- }
- }()
- if len(deferred) == 0 {
- return nil
- }
- applyCtx, cleanup := sc.trieCtxFactory(ctx)
- if cleanup != nil {
- defer cleanup()
- }
- if applyCtx == nil {
- return errors.New("StreamingCommitter: trieCtxFactory returned nil context for deferred apply")
- }
- if err := applyDeferredGuarded(applyCtx, deferred, sc.numWorkers); err != nil {
- return fmt.Errorf("apply deferred branch updates: %w", err)
- }
- return nil
-}
-
-// applyDeferredGuarded applies deferred branch updates, pre-merging in memory any
-// prefix emitted by more than one fold set because the apply context may be
-// write-only and a colliding update cannot re-read its predecessor from ctx.
-func applyDeferredGuarded(ctx PatriciaContext, deferred []*DeferredBranchUpdate, numWorkers int) error {
- if !hasDuplicatePrefix(deferred) {
- _, err := ApplyDeferredBranchUpdates(deferred, numWorkers, ctx.PutBranch)
- return err
- }
-
- merger := workerMergerPool.Get().(*BranchMerger)
- defer workerMergerPool.Put(merger)
-
- applied := make(map[string][]byte, len(deferred))
- for _, upd := range deferred {
- if upd == nil {
- continue
- }
- key := string(upd.prefix)
- if prev, ok := applied[key]; ok {
- upd.prev = bytes.Clone(prev)
- } else {
- prev, _, err := ctx.Branch(upd.prefix)
- if err != nil {
- return err
- }
- upd.prev = bytes.Clone(prev)
- }
- if err := mergeDeferredUpdate(upd, merger); err != nil {
- return err
- }
- if upd.encoded == nil {
- applied[key] = upd.prev
- continue
- }
- if err := ctx.PutBranch(upd.prefix, upd.encoded, upd.prev); err != nil {
- return err
- }
- applied[key] = bytes.Clone(upd.encoded)
- }
- return nil
-}
-
-func hasDuplicatePrefix(deferred []*DeferredBranchUpdate) bool {
- seen := make(map[string]struct{}, len(deferred))
- for _, upd := range deferred {
- if upd == nil {
- continue
- }
- key := string(upd.prefix)
- if _, ok := seen[key]; ok {
- return true
- }
- seen[key] = struct{}{}
- }
- return false
-}
-
-// Reset clears per-split state, the prefix trie, and staged deferred updates so the
-// committer can be reused for the next block.
-func (sc *StreamingCommitter) Reset() {
- sc.Stop()
- sc.releaseBase()
- if sc.trie != nil {
- sc.trie.Reset()
- }
- sc.dropSplitDeferred()
- clear(sc.splits)
- for _, upd := range sc.deferredForCaller {
- putDeferredUpdate(upd)
- }
- sc.deferredForCaller = nil
- sc.rootValid, sc.rootSeeded = false, false
-}
-
-// releaseBase drops the scheduler's persistent base and its context.
-func (sc *StreamingCommitter) releaseBase() {
- if sc.baseCleanup != nil {
- sc.baseCleanup()
- sc.baseCleanup = nil
- }
- sc.base = nil
-}
-
-// Release drops all owned state; the committer must not be used afterwards.
-// Repeat calls are safe no-ops.
-func (sc *StreamingCommitter) Release() {
- sc.Stop()
- sc.releaseBase()
- sc.dropSplitDeferred()
- sc.trie = nil
- sc.splits = nil
- for _, upd := range sc.deferredForCaller {
- putDeferredUpdate(upd)
- }
- sc.deferredForCaller = nil
- sc.rootValid, sc.rootSeeded = false, false
-}
diff --git a/execution/commitment/streaming_commitment_test.go b/execution/commitment/streaming_commitment_test.go
deleted file mode 100644
index 6a6ef49bb50..00000000000
--- a/execution/commitment/streaming_commitment_test.go
+++ /dev/null
@@ -1,995 +0,0 @@
-// Copyright 2026 The Erigon Authors
-// This file is part of Erigon.
-//
-// Erigon is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// Erigon is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with Erigon. If not, see .
-
-package commitment
-
-import (
- "bytes"
- "context"
- "encoding/binary"
- "math/rand"
- "sync"
- "testing"
- "time"
-
- "github.com/stretchr/testify/require"
-
- "github.com/erigontech/erigon/common"
- "github.com/erigontech/erigon/common/length"
-)
-
-type recordingStreamSink struct{ touches int }
-
-func (r *recordingStreamSink) TouchKey(hashedKey, plainKey []byte, update *Update) {
- r.touches++
-}
-
-func TestUpdates_NewEmpty_PreservesStreaming(t *testing.T) {
- u := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
- sink := &recordingStreamSink{}
- u.SetStreamingCommitter(sink)
-
- rotated := u.NewEmpty()
- require.True(t, rotated.Streaming(), "NewEmpty dropped the streaming funnel")
-
- addr := make([]byte, length.Addr)
- addr[0] = 0xab
- rotated.TouchPlainKeyDirect(string(addr), &Update{Flags: BalanceUpdate})
- require.NotZero(t, sink.touches, "rotated buffer did not forward touch to the streamer")
-}
-
-func streamingRoot(t *testing.T, workers int, keys [][]byte, upds []Update, idxOrder []int) ([]byte, *MockState) {
- t.Helper()
- sc, ms := newStreamingFixture(t, keys, upds, workers)
- defer sc.Release()
- for _, i := range idxOrder {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- return root, ms
-}
-
-func sequentialRoot(t *testing.T, keys [][]byte, upds []Update) ([]byte, *MockState) {
- t.Helper()
- return engineRoot(t, modeSeq, 0, keys, upds)
-}
-
-func TestStreaming_RandomOrderParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(99, 6000)
-
- idx := make([]int, len(keys))
- for i := range idx {
- idx[i] = i
- }
- rnd := rand.New(rand.NewSource(0xBEEF))
- rnd.Shuffle(len(idx), func(i, j int) { idx[i], idx[j] = idx[j], idx[i] })
-
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
- for _, w := range benchWorkerCounts() {
- strRoot, strMs := streamingRoot(t, w, keys, upds, idx)
- require.Equalf(t, seqRoot, strRoot, "streaming(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, strMs)
- }
-}
-
-func TestStreaming_DeepBranchParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildWhaleCorpus(bigAccountWhale(15_000))
-
- idx := make([]int, len(keys))
- for i := range idx {
- idx[i] = i
- }
- rnd := rand.New(rand.NewSource(0xD00D))
- rnd.Shuffle(len(idx), func(i, j int) { idx[i], idx[j] = idx[j], idx[i] })
-
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
- for _, w := range benchWorkerCounts() {
- strRoot, strMs := streamingRoot(t, w, keys, upds, idx)
- require.Equalf(t, seqRoot, strRoot, "streaming(workers=%d) deep root != sequential", w)
- requireBranchParity(t, seqMs, strMs)
- }
-}
-
-// Corpus must be collapse-free: repeated re-folds over a non-empty pre-image only stay parity-clean without collapses.
-func TestStreaming_NonEmptyPrevRefold(t *testing.T) {
- t.Parallel()
- const workers = 4
- ctx := context.Background()
- k1, u1 := genRandomAccountsStorage(400)
- k2, u2 := sparseBatch2(k1, 3, false)
-
- seqRoot, seqMs := runIncremental(t, modeSeq, 0, k1, u1, k2, u2)
-
- sc1, ms := newStreamingFixture(t, k1, u1, workers)
- touchAll(sc1, k1)
- _, err := sc1.Process(ctx)
- require.NoError(t, err)
- sc1.Release()
- require.NotEmpty(t, ms.cm)
- snap := snapshotBranches(ms)
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- sc2 := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- defer sc2.Release()
- sc2.SetNumWorkers(workers)
- touchAll(sc2, k2)
- for range 4 {
- require.NoError(t, sc2.foldDirtySplits(ctx))
- requireBranchesUnchanged(t, snap, ms)
- }
- root2, err := sc2.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot, root2, "streaming block-2 root after re-folds != sequential")
- requireBranchParity(t, seqMs, ms)
-}
-
-// touched nibbles get hash bytes seeded from seed so each update's cells are source-distinguishable in dedup assertions.
-func makeBranch(prefix []byte, afterMap uint16, touched []int, seed byte, prev []byte) *DeferredBranchUpdate {
- var cells [16]cellEncodeData
- var tm uint16
- for _, n := range touched {
- tm |= uint16(1) << uint(n)
- cells[n].hashLen = 32
- for b := range cells[n].hash {
- cells[n].hash[b] = seed + byte(n)
- }
- }
- raw, err := NewBranchEncoder(64).EncodeBranch(tm, tm, afterMap, &cells)
- if err != nil {
- panic(err)
- }
- return getDeferredUpdate(prefix, raw, prev)
-}
-
-// Settle condition is idle, not all-clean: the coalescing gate may legitimately leave a split dirty.
-func waitSchedulerIdle(t *testing.T, sc *StreamingCommitter) {
- t.Helper()
- deadline := time.Now().Add(15 * time.Second)
- stable := 0
- for {
- queued := 0
- sc.trieMu.RLock()
- for _, s := range sc.splits {
- s.mu.Lock()
- if s.queued {
- queued++
- }
- s.mu.Unlock()
- }
- sc.trieMu.RUnlock()
- if queued == 0 && len(sc.dirtyCh) == 0 && sc.inFlight.Load() == 0 {
- if stable++; stable >= 5 {
- return
- }
- } else {
- stable = 0
- }
- if time.Now().After(deadline) {
- t.Fatalf("scheduler did not go idle: queued=%d dirtyCh=%d", queued, len(sc.dirtyCh))
- }
- time.Sleep(time.Millisecond)
- }
-}
-
-func TestStreaming_SchedulerConcurrentParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(77, 4000)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- for _, w := range benchWorkerCounts() {
- sc, ms := newStreamingFixture(t, keys, upds, w, true)
-
- const goroutines = 4
- var wg sync.WaitGroup
- for g := range goroutines {
- wg.Go(func() {
- for i := g; i < len(keys); i += goroutines {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- })
- }
- wg.Wait()
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equalf(t, seqRoot, root, "scheduler(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, ms)
- sc.Release()
- }
-}
-
-func TestStreaming_StorageMidAccountFold(t *testing.T) {
- t.Parallel()
-
- t.Run("retouch_after_fold", func(t *testing.T) {
- keys, upds := buildMixedCorpus(33, 2500)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- sc, ms := newStreamingFixture(t, keys, upds, 4)
- defer sc.Release()
- sc.SetEagerFold(1) // below production floor: forces the gate path
- require.NoError(t, sc.StartScheduler(context.Background()))
-
- const hold = 8
- for i := 0; i < len(keys)-hold; i++ {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- waitSchedulerIdle(t, sc)
-
- for i := len(keys) - hold; i < len(keys); i++ {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- waitSchedulerIdle(t, sc)
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "streaming re-touch root != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-
- t.Run("mid_account_fold", func(t *testing.T) {
- keys, upds := genRandomAccountsStorage(300)
-
- withheld := -1
- for i, k := range keys {
- if len(k) > length.Addr {
- withheld = i
- break
- }
- }
- require.GreaterOrEqual(t, withheld, 0, "corpus must contain a storage key")
- targetNib := KeyToHexNibbleHash(keys[withheld])[0]
-
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- sc, ms := newStreamingFixture(t, keys, upds, 4)
- defer sc.Release()
- sc.SetEagerFold(1) // below production floor so the target split folds and foldGate fires
-
- var once sync.Once
- sc.SetFoldGate(func(nib byte) {
- if nib != targetNib {
- return
- }
- once.Do(func() {
- sc.TouchKey(KeyToHexNibbleHash(keys[withheld]), keys[withheld], nil)
- })
- })
-
- require.NoError(t, sc.StartScheduler(context.Background()))
- for i, k := range keys {
- if i == withheld {
- continue
- }
- sc.TouchKey(KeyToHexNibbleHash(k), k, nil)
- }
- waitSchedulerIdle(t, sc)
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "streaming storage-mid-fold root != sequential")
- requireBranchParity(t, seqMs, ms)
- require.Positive(t, sc.RefoldCount(), "expected a mid-fold re-fold to be triggered")
- })
-}
-
-func TestStreaming_SchedulerCollapseParity(t *testing.T) {
- t.Parallel()
- ctx := context.Background()
- k1, u1 := genRandomAccountsStorage(400)
- k2, u2 := sparseBatch2(k1, 3, true)
- seqRoot, seqMs := runIncremental(t, modeSeq, 0, k1, u1, k2, u2)
-
- for _, w := range []int{1, 4} {
- sc1, ms := newStreamingFixture(t, k1, u1, w)
- touchAll(sc1, k1)
- _, err := sc1.Process(ctx)
- require.NoError(t, err)
- sc1.Release()
- snap := snapshotBranches(ms)
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- sc2 := NewStreamingCommitter(mockTrieCtxFactory(ms), length.Addr, DefaultTrieConfig())
- sc2.SetNumWorkers(w)
- require.NoError(t, sc2.StartScheduler(ctx))
- touchAll(sc2, k2)
- sc2.Stop()
- requireBranchesUnchanged(t, snap, ms)
-
- root, err := sc2.Process(ctx)
- require.NoError(t, err)
- sc2.Release()
- require.Equalf(t, seqRoot, root, "scheduler collapse(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, ms)
- }
-}
-
-func TestStreaming_SplitMergeCollisionDedup(t *testing.T) {
- t.Parallel()
- prefix := []byte{0x0a, 0x03}
- all := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
-
- storePrev := func(ms *MockState) []byte {
- _, err := ApplyDeferredBranchUpdates(
- []*DeferredBranchUpdate{makeBranch(prefix, 0xFFFF, all, 0x10, nil)},
- 1, ms.PutBranch)
- require.NoError(t, err)
- return append([]byte(nil), ms.cm[string(prefix)]...)
- }
-
- guardMs := NewMockState(t)
- gp := storePrev(guardMs)
- require.NoError(t, applyDeferredGuarded(guardMs, []*DeferredBranchUpdate{
- makeBranch(prefix, 0xFFFF, []int{0, 1, 2, 3}, 0x40, gp),
- makeBranch(prefix, 0xFFFF, []int{4, 5, 6, 7}, 0x80, gp),
- }, 4))
- _, _, grow, err := BranchData(guardMs.cm[string(prefix)]).decodeCells()
- require.NoError(t, err)
- require.Equal(t, byte(0x40+0), grow[0].hash[0], "guard kept the split set's low-half cell")
- require.Equal(t, byte(0x80+4), grow[4].hash[0], "guard kept the merge set's high-half cell")
- require.Equal(t, byte(0x10+8), grow[8].hash[0], "guard kept prev's untouched cell")
-
- bareMs := NewMockState(t)
- bp := storePrev(bareMs)
- _, err = ApplyDeferredBranchUpdates([]*DeferredBranchUpdate{
- makeBranch(prefix, 0xFFFF, []int{0, 1, 2, 3}, 0x40, bp),
- makeBranch(prefix, 0xFFFF, []int{4, 5, 6, 7}, 0x80, bp),
- }, 4, bareMs.PutBranch)
- require.NoError(t, err)
- _, _, brow, err := BranchData(bareMs.cm[string(prefix)]).decodeCells()
- require.NoError(t, err)
- require.Equal(t, byte(0x10+0), brow[0].hash[0], "bare apply dropped the split set's low half (clobbered by prev)")
- require.Equal(t, byte(0x80+4), brow[4].hash[0], "bare apply kept the merge set's high half")
-}
-
-// Models the production write-only apply context: PutBranch buffers writes that Branch cannot read until drained.
-type drainContext struct {
- *MockState
- pending map[string][]byte
-}
-
-func newDrainContext(ms *MockState) *drainContext {
- return &drainContext{MockState: ms, pending: make(map[string][]byte)}
-}
-
-func (d *drainContext) PutBranch(prefix, data, _ []byte) error {
- d.pending[string(prefix)] = bytes.Clone(data)
- return nil
-}
-
-func (d *drainContext) drain() error {
- for k, v := range d.pending {
- if err := d.MockState.PutBranch([]byte(k), v, nil); err != nil {
- return err
- }
- }
- return nil
-}
-
-func TestStreaming_SplitMergeCollisionDedup_WriteOnlyCtx(t *testing.T) {
- t.Parallel()
- prefix := []byte{0x0a, 0x03}
- all := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
-
- ms := NewMockState(t)
- _, err := ApplyDeferredBranchUpdates(
- []*DeferredBranchUpdate{makeBranch(prefix, 0xFFFF, all, 0x10, nil)},
- 1, ms.PutBranch)
- require.NoError(t, err)
- prev := append([]byte(nil), ms.cm[string(prefix)]...)
-
- dctx := newDrainContext(ms)
- require.NoError(t, applyDeferredGuarded(dctx, []*DeferredBranchUpdate{
- makeBranch(prefix, 0xFFFF, []int{0, 1, 2, 3}, 0x40, prev),
- makeBranch(prefix, 0xFFFF, []int{4, 5, 6, 7}, 0x80, prev),
- }, 4))
- require.NoError(t, dctx.drain())
-
- _, _, row, err := BranchData(ms.cm[string(prefix)]).decodeCells()
- require.NoError(t, err)
- require.Equal(t, byte(0x40+0), row[0].hash[0], "split set's low-half cell survived write-only apply")
- require.Equal(t, byte(0x80+4), row[4].hash[0], "merge set's high-half cell survived write-only apply")
- require.Equal(t, byte(0x10+8), row[8].hash[0], "prev's untouched cell survived")
-}
-
-func foldedSplitCount(sc *StreamingCommitter) int {
- n := 0
- sc.trieMu.RLock()
- for _, s := range sc.splits {
- s.mu.Lock()
- if s.folded {
- n++
- }
- s.mu.Unlock()
- }
- sc.trieMu.RUnlock()
- return n
-}
-
-func splitCount(sc *StreamingCommitter) int {
- sc.trieMu.RLock()
- n := len(sc.splits)
- sc.trieMu.RUnlock()
- return n
-}
-
-func TestStreaming_FoldEagerPolicy(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(123, 5000)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- newCommitter := func() (*StreamingCommitter, *MockState) {
- sc, ms := newStreamingFixture(t, keys, upds, 4)
- sc.SetEagerFold(1) // below production floor: forces the eager path
- return sc, ms
- }
-
- t.Run("lazy_fall_through", func(t *testing.T) {
- sc, ms := newCommitter()
- defer sc.Release()
- touchAll(sc, keys)
- require.Zero(t, foldedSplitCount(sc), "no scheduler: nothing folds before Process")
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "lazy fall-through root != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-
- t.Run("eager_drained", func(t *testing.T) {
- sc, ms := newCommitter()
- defer sc.Release()
- require.NoError(t, sc.StartScheduler(context.Background()))
- touchAll(sc, keys)
- waitSchedulerIdle(t, sc)
- require.Positive(t, foldedSplitCount(sc), "eager policy must fold splits in the background")
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "eager drained root != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-
- t.Run("eager_partial_fall_through", func(t *testing.T) {
- sc, ms := newCommitter()
- defer sc.Release()
- require.NoError(t, sc.StartScheduler(context.Background()))
- touchAll(sc, keys)
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "eager partial fall-through root != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-}
-
-// carried=true touches via TouchPlainKeyDirect (carried *Update); carried=false via TouchPlainKey (nil/ctx-read).
-func streamingViaUpdatesRoot(t *testing.T, workers int, keys [][]byte, upds []Update, carried bool) ([]byte, *MockState) {
- t.Helper()
- sc, ms := newStreamingFixture(t, keys, upds, workers)
- defer sc.Release()
-
- ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
- defer ut.Close()
- ut.SetStreamingCommitter(sc)
- require.True(t, ut.Streaming())
-
- for i, key := range keys {
- if carried {
- ut.TouchPlainKeyDirect(string(key), &upds[i])
- } else {
- ut.TouchPlainKey(string(key), nil, ut.TouchAccount)
- }
- }
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- return root, ms
-}
-
-func TestStreaming_UpdatesFunnelParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(7, 6000)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- for _, carried := range []bool{false, true} {
- for _, w := range benchWorkerCounts() {
- root, ms := streamingViaUpdatesRoot(t, w, keys, upds, carried)
- require.Equalf(t, seqRoot, root, "funnel(carried=%v,workers=%d) root != sequential", carried, w)
- requireBranchParity(t, seqMs, ms)
- }
- }
-}
-
-func TestStreaming_UpdatesLifetimeRegression(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(31, 4000)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- sc, ms := newStreamingFixture(t, keys, upds, 4)
- defer sc.Release()
-
- ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash)
- defer ut.Close()
- ut.SetStreamingCommitter(sc)
-
- var held []*Update
- for i, key := range keys {
- kb := append([]byte(nil), key...)
- ub := upds[i]
- ut.TouchPlainKeyDirect(common.ToStringZeroCopy(kb), &ub)
- // Corrupt the caller's key backing after Touch: a non-copying intern would fold the corruption.
- for j := range kb {
- kb[j] ^= 0xFF
- }
- held = append(held, &ub)
- }
- // Clobber every caller-owned Update before folding: a non-copying branch would fold these deletes.
- for _, u := range held {
- *u = Update{Flags: DeleteUpdate}
- }
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equal(t, seqRoot, root, "mutating caller buffers after Touch changed the root")
- requireBranchParity(t, seqMs, ms)
-}
-
-func TestInitializeTrieAndUpdates_StreamingVariant(t *testing.T) {
- t.Parallel()
-
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, upd := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer upd.Close()
- defer trie.Release()
-
- require.IsType(t, (*ParallelPatriciaHashed)(nil), trie)
- require.Equal(t, VariantStreamingHexPatricia, trie.Variant())
- pt := trie.(*ParallelPatriciaHashed)
- require.NotNil(t, pt.streaming, "streaming variant must attach a StreamingCommitter")
- require.Equal(t, ModeParallel, upd.Mode())
- require.NotNil(t, upd.parallel)
- require.True(t, upd.Streaming(), "Updates must forward touches to the committer")
-}
-
-func streamingViaPublicProcessRoot(t *testing.T, workers int, keys [][]byte, upds []Update) ([]byte, *MockState) {
- t.Helper()
- return engineRoot(t, modeStreamingPublic, workers, keys, upds)
-}
-
-func TestStreaming_PublicProcessParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildWhaleCorpus(bigAccountWhale(15_000))
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- for _, w := range benchWorkerCounts() {
- root, ms := streamingViaPublicProcessRoot(t, w, keys, upds)
- require.Equalf(t, seqRoot, root, "public Process(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, ms)
- }
-}
-
-func requireResetClean(t *testing.T, sc *StreamingCommitter) {
- t.Helper()
- require.Empty(t, sc.splits, "Reset left stale split state")
- require.Nil(t, sc.deferredForCaller, "Reset left staged deferred updates")
- require.Nil(t, sc.base, "Reset left the scheduler base alive")
- require.False(t, sc.started.Load(), "Reset left the scheduler running")
- require.NotNil(t, sc.trie, "Reset must keep a usable prefix trie")
- require.Zero(t, sc.trie.root.subtreeCount, "Reset left prefix-trie entries")
-}
-
-func TestStreaming_NewSplitMidBlock(t *testing.T) {
- t.Parallel()
- keys, upds := buildMixedCorpus(91, 4000)
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- var early, late []int
- for i, k := range keys {
- if KeyToHexNibbleHash(k)[0] < 8 {
- early = append(early, i)
- } else {
- late = append(late, i)
- }
- }
- require.NotEmpty(t, early, "corpus must populate the lower top-nibble half")
- require.NotEmpty(t, late, "corpus must populate the upper top-nibble half")
-
- for _, w := range benchWorkerCounts() {
- sc, ms := newStreamingFixture(t, keys, upds, w)
- sc.SetEagerFold(1) // below production floor: forces background folding
- require.NoError(t, sc.StartScheduler(context.Background()))
-
- for _, i := range early {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- waitSchedulerIdle(t, sc)
- earlySplits := splitCount(sc)
- require.Positive(t, earlySplits, "early touches must create the lower-half splits")
-
- for _, i := range late {
- sc.TouchKey(KeyToHexNibbleHash(keys[i]), keys[i], nil)
- }
- waitSchedulerIdle(t, sc)
- require.Greater(t, splitCount(sc), earlySplits, "late touches must create brand-new top-nibble splits")
-
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
- require.Equalf(t, seqRoot, root, "new-split-mid-block(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, ms)
- sc.Release()
- }
-}
-
-func TestStreaming_MultiBlockReuse(t *testing.T) {
- t.Parallel()
-
- t.Run("reset", func(t *testing.T) {
- ctx := context.Background()
- k1, u1 := genRandomAccountsStorage(400)
- k2, u2 := sparseBatch2(k1, 3, true)
-
- seqRoot1, _ := sequentialRoot(t, k1, u1)
- seqRoot2, seqMs := runIncremental(t, modeSeq, 0, k1, u1, k2, u2)
-
- sc, ms := newStreamingFixture(t, k1, u1, 4, true)
- defer sc.Release()
-
- touchAll(sc, k1)
- root1, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot1, root1, "block-1 streaming root != sequential")
-
- sc.Reset()
- requireResetClean(t, sc)
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- require.NoError(t, sc.StartScheduler(ctx))
- touchAll(sc, k2)
- root2, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot2, root2, "block-2 streaming root after reset != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-
- t.Run("no_reset", func(t *testing.T) {
- ctx := context.Background()
- k1, u1 := genRandomAccountsStorage(400)
- k2, u2 := sparseBatch2(k1, 3, true)
-
- seqRoot1, _ := sequentialRoot(t, k1, u1)
- seqRoot2, seqMs := runIncremental(t, modeSeq, 0, k1, u1, k2, u2)
-
- sc, ms := newStreamingFixture(t, k1, u1, 4)
- defer sc.Release()
-
- touchAll(sc, k1)
- root1, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot1, root1, "block-1 streaming root != sequential")
-
- require.Zero(t, sc.trie.root.subtreeCount, "Process left block-1 keys in the prefix trie")
- require.Empty(t, sc.splits, "Process left stale split state")
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- touchAll(sc, k2)
- root2, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot2, root2, "block-2 streaming root without reset != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-
- t.Run("scheduler_then_no_reset", func(t *testing.T) {
- ctx := context.Background()
- k1, u1 := genRandomAccountsStorage(400)
- k2, u2 := sparseBatch2(k1, 3, true)
-
- seqRoot1, _ := sequentialRoot(t, k1, u1)
- seqRoot2, seqMs := runIncremental(t, modeSeq, 0, k1, u1, k2, u2)
-
- sc, ms := newStreamingFixture(t, k1, u1, 4, true)
- defer sc.Release()
-
- touchAll(sc, k1)
- root1, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot1, root1, "block-1 scheduler root != sequential")
- require.Nil(t, sc.base, "Process must release the scheduler base after folding it down")
-
- require.NoError(t, ms.applyPlainUpdates(k2, u2))
- touchAll(sc, k2)
- root2, err := sc.Process(ctx)
- require.NoError(t, err)
- require.Equal(t, seqRoot2, root2, "block-2 lazy root after scheduler block (no reset) != sequential")
- requireBranchParity(t, seqMs, ms)
- })
-}
-
-func TestStreamingCommitterStateRoundTrip(t *testing.T) {
- t.Parallel()
-
- plainKeys, updates := NewUpdateBuilder().
- Balance("68ee6c0e9cdc73b2b2d52dbd79f19d24fe25e2f9", 42).
- Balance("a1b2c3d4e5f60718293a4b5c6d7e8f9012345678", 7).
- Balance("ffeeddccbbaa00112233445566778899aabbccdd", 99).
- Build()
-
- ms := NewMockState(t)
- ms.SetConcurrentCommitment(true)
- require.NoError(t, ms.applyPlainUpdates(plainKeys, updates))
-
- cfg := DefaultTrieConfig()
- cfg.Variant = VariantStreamingHexPatricia
- trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut.Close()
- defer trie.Release()
-
- pt := trie.(*ParallelPatriciaHashed)
- pt.SetNumWorkers(1)
- pt.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt.ResetContext(ms)
-
- for _, key := range plainKeys {
- ut.TouchPlainKey(string(key), nil, ut.TouchAccount)
- }
- published, err := trie.Process(context.Background(), ut, "", nil, WarmupConfig{})
- require.NoError(t, err)
- require.NotEmpty(t, published)
-
- tmpl := pt.RootTrie()
- require.True(t, tmpl.rootChecked, "streaming template.rootChecked must be promoted from the committer")
- require.True(t, tmpl.rootTouched, "streaming template.rootTouched must be promoted from the committer")
- require.True(t, tmpl.rootPresent, "streaming template.rootPresent must be promoted from the committer")
-
- encoded, err := tmpl.EncodeCurrentState(nil)
- require.NoError(t, err)
- require.NotEmpty(t, encoded, "EncodeCurrentState must capture template state mirrored from the committer")
-
- trie2, ut2 := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg)
- defer ut2.Close()
- defer trie2.Release()
- pt2 := trie2.(*ParallelPatriciaHashed)
- pt2.SetTrieContextFactory(mockTrieCtxFactory(ms))
- pt2.ResetContext(ms)
- require.NoError(t, pt2.RootTrie().SetState(encoded))
-
- restored, err := pt2.RootHash()
- require.NoError(t, err)
- require.Equal(t, published, restored,
- "RootHash after SetState must reproduce the published streaming root")
-}
-
-func TestKeyArena_PointerStability(t *testing.T) {
- var arena keyArena
-
- inputs := make([][]byte, 0, 4096)
- got := make([][]byte, 0, 4096)
- // 4096 small keys roll the arena over at least two chunks.
- for i := range 4096 {
- in := bytes.Repeat([]byte{byte(i), byte(i >> 8)}, 32)
- inputs = append(inputs, in)
- got = append(got, arena.copy(in))
- }
- // Oversized key forces the max(keyArenaChunk, len) allocation path.
- big := bytes.Repeat([]byte{0xAB}, keyArenaChunk+128)
- inputs = append(inputs, big)
- got = append(got, arena.copy(big))
-
- for i, in := range inputs {
- require.True(t, bytes.Equal(in, got[i]),
- "key %d corrupted: returned slice does not equal its input", i)
- require.Equal(t, len(got[i]), cap(got[i]),
- "key %d not full-cap: a caller append could overwrite the next key", i)
- }
-
- for i := range got {
- for j := range got[i] {
- got[i][j] = byte(i)
- }
- }
- for i := range got {
- for j := range got[i] {
- require.Equal(t, byte(i), got[i][j],
- "key %d overlaps another arena slice (overwritten at byte %d)", i, j)
- }
- }
-}
-
-// Mixes shallow account forks with a whale storage subtree forking below depth 64 to fan out folds across many depths.
-func buildMultiDepthCorpus() (keys [][]byte, upds []Update) {
- mk, mu := buildMixedCorpus(0xD15C0DE, 6000)
- _, _, _, _, pk, pu, _ := whaleByNibble(20_000)
- keys = append(keys, mk...)
- keys = append(keys, pk...)
- upds = append(upds, mu...)
- upds = append(upds, pu...)
- return keys, upds
-}
-
-func parallelRoot(t *testing.T, workers int, keys [][]byte, upds []Update) ([]byte, *MockState) {
- t.Helper()
- return engineRoot(t, modeParallel, workers, keys, upds)
-}
-
-func TestStreaming_MultiDepthSplitParity(t *testing.T) {
- t.Parallel()
- keys, upds := buildMultiDepthCorpus()
-
- seqRoot, seqMs := sequentialRoot(t, keys, upds)
-
- parRoot, parMs := parallelRoot(t, 4, keys, upds)
- require.Equal(t, seqRoot, parRoot, "parallel root != sequential")
- requireBranchParity(t, seqMs, parMs)
-
- for _, w := range benchWorkerCounts() {
- sc, ms := newStreamingFixture(t, keys, upds, w)
- touchAll(sc, keys)
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
-
- require.Equalf(t, seqRoot, root, "multi-depth streaming(workers=%d) root != ModeDirect", w)
- require.Equalf(t, parRoot, root, "multi-depth streaming(workers=%d) root != ModeParallel", w)
- requireBranchParity(t, seqMs, ms)
- // First-commit whale takes the streaming-recursion fallback, not the deep fold.
- sc.Release()
- }
-}
-
-// Embedding the collapsing whale among many accounts is load-bearing: a single-account trie yields a degenerate collapse root.
-func TestStreaming_MultiDepthCollapseParity(t *testing.T) {
- t.Parallel()
- wk1, wu1, wk2, wu2 := whaleCollapseCorpus()
- for _, tc := range []struct {
- name string
- mixSeed int64
- mixKeys int
- }{
- {"embed_c0ffee_4000", 0xC0FFEE, 4000},
- {"embed_5eed_3000", 0x5EED, 3000},
- } {
- t.Run(tc.name, func(t *testing.T) {
- mk, mu := buildMixedCorpus(tc.mixSeed, tc.mixKeys)
- k1 := append(append([][]byte{}, mk...), wk1...)
- u1 := append(append([]Update{}, mu...), wu1...)
-
- seqRoot, seqMs := runIncremental(t, modeSeq, 0, k1, u1, wk2, wu2)
- for _, w := range benchWorkerCounts() {
- requireIncrementalEquiv(t, k1, u1, wk2, wu2, w)
- strRoot, strMs := runIncremental(t, modeStreaming, w, k1, u1, wk2, wu2)
- require.Equalf(t, seqRoot, strRoot, "whale storage collapse(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, strMs)
- }
- })
- }
-}
-
-func TestStreaming_FullCollapseParity(t *testing.T) {
- t.Parallel()
- wk1, wu1, wk2, wu2 := whaleFullCollapseCorpus()
- mk, mu := buildMixedCorpus(0xC0FFEE, 4000)
- k1 := append(append([][]byte{}, mk...), wk1...)
- u1 := append(append([]Update{}, mu...), wu1...)
- for _, w := range benchWorkerCounts() {
- requireIncrementalEquiv(t, k1, u1, wk2, wu2, w)
- }
-}
-
-func TestStreaming_StorageInteriorSplits(t *testing.T) {
- t.Parallel()
- _, _, _, _, pk, upds, _ := whaleByNibble(20_000)
-
- seqRoot, seqMs := sequentialRoot(t, pk, upds)
-
- for _, w := range benchWorkerCounts() {
- sc, ms := newStreamingFixture(t, pk, upds, w)
- touchAll(sc, pk)
- root, err := sc.Process(context.Background())
- require.NoError(t, err)
-
- require.Equalf(t, seqRoot, root, "whale storage-interior split(workers=%d) root != sequential", w)
- requireBranchParity(t, seqMs, ms)
- // First-commit whale takes the streaming-recursion fallback, not the deep fold.
- sc.Release()
- }
-}
-
-// Block 2 leaves a third of slots untouched on disk: the untouched-sibling read forces a mid-fold self-flush.
-func whaleCollapseCorpus() (pk [][]byte, upds []Update, k2 [][]byte, u2 []Update) {
- var addr []byte
- addr, _, _, _, pk, upds, _ = whaleByNibble(30_000)
-
- k2 = [][]byte{addr}
- u2 = []Update{{Flags: BalanceUpdate | NonceUpdate}}
- u2[0].Balance.SetUint64(99)
- u2[0].Nonce = 7
- for i := range pk {
- if len(pk[i]) == length.Addr || i%3 == 2 {
- continue
- }
- var u Update
- if i%3 == 0 {
- u.Flags = DeleteUpdate
- } else {
- u.Flags = StorageUpdate
- u.StorageLen = 4
- binary.BigEndian.PutUint32(u.Storage[:4], uint32(i)*97+3)
- }
- k2 = append(k2, pk[i])
- u2 = append(u2, u)
- }
- return pk, upds, k2, u2
-}
-
-// Deleting every storage slot exercises the all-children-collapsed path, which must yield the empty-trie root, not a zero hash.
-func whaleFullCollapseCorpus() (pk [][]byte, upds []Update, k2 [][]byte, u2 []Update) {
- var addr []byte
- addr, _, _, _, pk, upds, _ = whaleByNibble(30_000)
-
- k2 = [][]byte{addr}
- u2 = []Update{{Flags: BalanceUpdate | NonceUpdate}}
- u2[0].Balance.SetUint64(99)
- u2[0].Nonce = 7
- for i := range pk {
- if len(pk[i]) == length.Addr {
- continue
- }
- k2 = append(k2, pk[i])
- u2 = append(u2, Update{Flags: DeleteUpdate})
- }
- return pk, upds, k2, u2
-}
-
-func TestKeyArena_ChunkSizedFromRemaining(t *testing.T) {
- const keyLen = 144
-
- t.Run("small subtree does not burn a full chunk", func(t *testing.T) {
- const keys = 32
- arena := keyArena{remaining: keys}
- for range keys {
- arena.copy(make([]byte, keyLen))
- }
- require.Equal(t, keys*keyLen, cap(arena.buf))
- })
-
- t.Run("large subtree still caps at one chunk", func(t *testing.T) {
- arena := keyArena{remaining: 10 * keyArenaChunk / keyLen}
- arena.copy(make([]byte, keyLen))
- require.Equal(t, keyArenaChunk, cap(arena.buf))
- })
-
- t.Run("oversized key gets its own backing", func(t *testing.T) {
- arena := keyArena{remaining: 4}
- got := arena.copy(make([]byte, 2*keyArenaChunk))
- require.Len(t, got, 2*keyArenaChunk)
- require.Equal(t, 2*keyArenaChunk, cap(arena.buf))
- })
-
- t.Run("copies stay stable across a chunk swap", func(t *testing.T) {
- arena := keyArena{remaining: 2}
- first := arena.copy(bytes.Repeat([]byte{0xAA}, keyLen))
- for i := range 8 {
- arena.copy(bytes.Repeat([]byte{byte(i)}, keyLen))
- }
- require.Equal(t, bytes.Repeat([]byte{0xAA}, keyLen), first)
- })
-
- t.Run("unhinted arena falls back to one key per chunk growth", func(t *testing.T) {
- var arena keyArena
- got := arena.copy(make([]byte, keyLen))
- require.Len(t, got, keyLen)
- require.Equal(t, keyLen, cap(arena.buf))
- })
-}
diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go
index 14c5bce4db0..dfeca3b0510 100644
--- a/execution/commitment/streaming_deep_fold.go
+++ b/execution/commitment/streaming_deep_fold.go
@@ -30,6 +30,14 @@ import (
"github.com/erigontech/erigon/execution/commitment/nibbles"
)
+// touchedKey is a snapshotted touch the mounted parallel fold replays; hk is
+// copied off the walk path while pk/upd reference the caller's stable backing.
+type touchedKey struct {
+ hk []byte
+ pk []byte
+ upd *Update
+}
+
// maxFoldConcurrency caps the whale-storage fold fan-out at the CPUs the process
// may run on. The account-mount and whale-storage fan-outs nest, so an unshared
// per-level limit of numWorkers permits ~numWorkers² runnable leaf goroutines; one
@@ -303,6 +311,28 @@ func newDeferredStorageWorker(ctx context.Context, accountKeyLen int16, cfg Trie
}
}
+// keyArena copies walk-path nibbles into chunked backing buffers so each
+// collected key gets a stable slice without one allocation per key. remaining
+// is the caller's expected key count; without it a subtree far smaller than a
+// chunk still burns a whole chunk, and most subtrees are.
+type keyArena struct {
+ buf []byte
+ remaining int
+}
+
+const keyArenaChunk = 64 * 1024
+
+func (a *keyArena) copy(hk []byte) []byte {
+ if len(hk) > cap(a.buf)-len(a.buf) {
+ want := len(hk) * max(a.remaining, 1)
+ a.buf = make([]byte, 0, max(min(want, keyArenaChunk), len(hk)))
+ }
+ a.remaining--
+ start := len(a.buf)
+ a.buf = append(a.buf, hk...)
+ return a.buf[start:len(a.buf):len(a.buf)]
+}
+
// collectSubtreeKeys walks a subtree in sorted order; it copies each key's hashed
// nibbles off the reused walk path but leaves plainKey/update aliased.
func collectSubtreeKeys(node *prefixNode, path []byte) []touchedKey {
diff --git a/node/cli/default_flags.go b/node/cli/default_flags.go
index 7512e900c4f..5baba4540ef 100644
--- a/node/cli/default_flags.go
+++ b/node/cli/default_flags.go
@@ -278,7 +278,6 @@ var DefaultFlags = []cli.Flag{
&utils.GDBMeFlag,
&utils.ExperimentalParallelCommitmentFlag,
- &utils.ExperimentalStreamingCommitmentFlag,
&utils.MCPDisableFlag,
&utils.MCPAddrFlag,
diff --git a/node/eth/backend.go b/node/eth/backend.go
index 81e951a4377..e66891a0cf4 100644
--- a/node/eth/backend.go
+++ b/node/eth/backend.go
@@ -309,9 +309,6 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger
if config.ExperimentalParallelCommitment {
statecfg.ExperimentalParallelCommitment = true
}
- if config.ExperimentalStreamingCommitment {
- statecfg.ExperimentalStreamingCommitment = true
- }
if err := stages.UpdateMetrics(tx); err != nil {
return err
diff --git a/node/ethconfig/config.go b/node/ethconfig/config.go
index b6eff166c21..0d1899904a9 100644
--- a/node/ethconfig/config.go
+++ b/node/ethconfig/config.go
@@ -319,12 +319,11 @@ type Sync struct {
LoopBlockLimit uint
ParallelStateFlushing bool
- ChaosMonkey bool
- AlwaysGenerateChangesets bool
- MaxReorgDepth uint64
- KeepExecutionProofs bool
- ExperimentalParallelCommitment bool
- ExperimentalStreamingCommitment bool
- PersistReceiptsCacheV2 bool
- SnapshotDownloadToBlock uint64 // exclusive [0,toBlock)
+ ChaosMonkey bool
+ AlwaysGenerateChangesets bool
+ MaxReorgDepth uint64
+ KeepExecutionProofs bool
+ ExperimentalParallelCommitment bool
+ PersistReceiptsCacheV2 bool
+ SnapshotDownloadToBlock uint64 // exclusive [0,toBlock)
}