From 0f17a3a939001eea3d1119040b20356480b14a46 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 11 Aug 2026 19:42:34 +0700 Subject: [PATCH 1/4] execution/commitment: remove the streaming commitment mode The streaming committer's background fold pool was never started in production: every StartScheduler caller was a test, so sc.base was always nil, sc.started always false, and TouchKey's enqueue branch never taken. --experimental.streaming-commitment advertised "overlaps folding with block execution" but folded synchronously, taking the same shape as the mounted parallel path through splits instead of mounts. Removes streaming_commitment.go, its tests, the StreamingCommitter wiring in ParallelPatriciaHashed and Updates, VariantStreamingHexPatricia, and the flag with its config plumbing. Kept, with live callers on the mounted parallel path: all of streaming_deep_fold.go (foldStorageRoot, dfsSubtreeDeep, unfoldStorageBase); keyArena/keyArenaChunk/touchedKey, moved there from streaming_commitment.go because collectSubtreeKeys needs them; stitchSplitCells, moved into parallel_mount.go. DeepLocalFolds moves to ParallelPatriciaHashed rather than disappearing with the committer. TestDeepFold_FreshWhaleFoldsParallel and TestDeepFold_ExistingWhaleStillDemotes assert on it to pin which fold path ran; without it they would degrade to root-parity checks that pass whichever path is taken. Mutation-checked: dropping the counter increment turns FreshWhale red. --- cmd/evm/staterunner_test.go | 3 - cmd/integration/commands/flags.go | 1 - cmd/utils/flags.go | 12 - db/state/execctx/commitment_flag_test.go | 57 -- db/state/execctx/domain_shared.go | 5 - db/state/squeeze.go | 5 +- db/state/statecfg/state_schema.go | 6 - execution/commitment/additive_updates_test.go | 21 - execution/commitment/commitment.go | 50 +- .../commitmentdb/commitment_context.go | 24 +- .../deepfold_emptystorage_regression_test.go | 2 - .../commitment/deepfold_regression_test.go | 2 - .../deepfold_retouch_regression_test.go | 2 - ...old_singleslot_reexpand_regression_test.go | 2 - .../deepfold_subset_regression_test.go | 31 +- .../mode_parallel_lifecycle_test.go | 105 -- execution/commitment/parallel_mount.go | 25 +- .../commitment/parallel_patricia_hashed.go | 79 +- .../parallel_patricia_hashed_test.go | 2 +- .../parallel_streaming_bench_test.go | 88 -- execution/commitment/parallel_testkit_test.go | 108 +- execution/commitment/parallel_trace_test.go | 2 +- .../state_roundtrip_regression_test.go | 95 -- execution/commitment/streaming_commitment.go | 955 ------------------ .../commitment/streaming_commitment_test.go | 953 ----------------- execution/commitment/streaming_deep_fold.go | 23 + node/cli/default_flags.go | 1 - node/eth/backend.go | 3 - node/ethconfig/config.go | 15 +- 29 files changed, 125 insertions(+), 2552 deletions(-) delete mode 100644 execution/commitment/streaming_commitment.go delete mode 100644 execution/commitment/streaming_commitment_test.go 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..76e197a081e 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 } @@ -110,57 +107,3 @@ func TestSharedDomains_ParallelFlag_RootEquivalence(t *testing.T) { 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..2c7c76070cc 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -291,11 +291,6 @@ type SharedDomains struct { // 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: return commitment.VariantParallelHexPatricia } diff --git a/db/state/squeeze.go b/db/state/squeeze.go index 5fc49944dd2..5512632d56c 100644 --- a/db/state/squeeze.go +++ b/db/state/squeeze.go @@ -1020,12 +1020,9 @@ 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: trieVariant = commitment.VariantParallelHexPatricia } @@ -1063,7 +1060,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 parallel { 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 c3c395edd98..f1832dff551 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_emptystorage_regression_test.go b/execution/commitment/deepfold_emptystorage_regression_test.go index 9fead49507e..39515253d20 100644 --- a/execution/commitment/deepfold_emptystorage_regression_test.go +++ b/execution/commitment/deepfold_emptystorage_regression_test.go @@ -83,8 +83,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) diff --git a/execution/commitment/deepfold_regression_test.go b/execution/commitment/deepfold_regression_test.go index 9a1c39da829..9f3e4bfa088 100644 --- a/execution/commitment/deepfold_regression_test.go +++ b/execution/commitment/deepfold_regression_test.go @@ -222,8 +222,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) diff --git a/execution/commitment/deepfold_retouch_regression_test.go b/execution/commitment/deepfold_retouch_regression_test.go index bd4e47bf256..4af8a9c31f3 100644 --- a/execution/commitment/deepfold_retouch_regression_test.go +++ b/execution/commitment/deepfold_retouch_regression_test.go @@ -85,8 +85,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) diff --git a/execution/commitment/deepfold_singleslot_reexpand_regression_test.go b/execution/commitment/deepfold_singleslot_reexpand_regression_test.go index ac1f0a51b3b..15152887995 100644 --- a/execution/commitment/deepfold_singleslot_reexpand_regression_test.go +++ b/execution/commitment/deepfold_singleslot_reexpand_regression_test.go @@ -103,8 +103,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) diff --git a/execution/commitment/deepfold_subset_regression_test.go b/execution/commitment/deepfold_subset_regression_test.go index bc84d14c568..5c7ff45c169 100644 --- a/execution/commitment/deepfold_subset_regression_test.go +++ b/execution/commitment/deepfold_subset_regression_test.go @@ -17,7 +17,6 @@ package commitment import ( - "context" "encoding/hex" "math/rand" "testing" @@ -113,7 +112,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) @@ -124,17 +123,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 @@ -149,16 +140,8 @@ 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") } 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 008e74b3d2b..879d340d732 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -77,6 +77,25 @@ func (hph *HexPatriciaHashed) mountTo(root *HexPatriciaHashed, nibble int) { } } +// 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 @@ -152,7 +171,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..d5a7428476a 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,6 @@ func (p *ParallelPatriciaHashed) Reset() { p.template.Reset() } p.rootHash.Store(nil) - if p.streaming != nil { - p.streaming.Reset() - } } // Release frees the template and worker pool; the instance must not be used afterwards. Repeat calls are no-ops. @@ -117,10 +112,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 +124,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 +135,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 +191,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 +207,6 @@ func (p *ParallelPatriciaHashed) EnableCsvMetrics(filePathPrefix string) { } func (p *ParallelPatriciaHashed) Variant() TrieVariant { - if p.streaming != nil { - return VariantStreamingHexPatricia - } return VariantParallelHexPatricia } @@ -250,26 +224,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 +243,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 +256,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..083cd09da1a 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 { @@ -196,39 +164,6 @@ func processModeBatchState(t *testing.T, ms *MockState, mode runMode, workers in }) } 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())) - } - for _, k := range keys { - sc.TouchKey(KeyToHexNibbleHash(k), k, 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()) default: tr := newSeqTrie(t, ms) defer tr.Release() @@ -239,6 +174,32 @@ 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() + 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] + }) + } + 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 +209,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 +245,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 44af52de98d..00000000000 --- a/execution/commitment/streaming_commitment.go +++ /dev/null @@ -1,955 +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. -type keyArena struct{ buf []byte } - -const keyArenaChunk = 64 * 1024 - -func (a *keyArena) copy(hk []byte) []byte { - if len(hk) > cap(a.buf)-len(a.buf) { - a.buf = make([]byte, 0, max(keyArenaChunk, len(hk))) - } - 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 c8b32a91786..00000000000 --- a/execution/commitment/streaming_commitment_test.go +++ /dev/null @@ -1,953 +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 -} diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go index de1cc4ed435..52763187941 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 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 +} + // 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,21 @@ 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. +type keyArena struct{ buf []byte } + +const keyArenaChunk = 64 * 1024 + +func (a *keyArena) copy(hk []byte) []byte { + if len(hk) > cap(a.buf)-len(a.buf) { + a.buf = make([]byte, 0, max(keyArenaChunk, len(hk))) + } + 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) } From 3d587c4ca1e4a92960b39653d0613c63b608e6a7 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 11 Aug 2026 19:55:18 +0700 Subject: [PATCH 2/4] db/state: collapse the single-case trie-variant switches left by the streaming removal --- db/state/execctx/commitment_flag_test.go | 1 - db/state/execctx/domain_shared.go | 3 +-- db/state/squeeze.go | 6 ++---- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/db/state/execctx/commitment_flag_test.go b/db/state/execctx/commitment_flag_test.go index 76e197a081e..241e913b97f 100644 --- a/db/state/execctx/commitment_flag_test.go +++ b/db/state/execctx/commitment_flag_test.go @@ -106,4 +106,3 @@ func TestSharedDomains_ParallelFlag_RootEquivalence(t *testing.T) { "sequential and parallel commitment roots must match: sequential=%x parallel=%x", seqRoot, parRoot) } - diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index 2c7c76070cc..071176eb652 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -290,8 +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 { - 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 5512632d56c..17df4b4ef9b 100644 --- a/db/state/squeeze.go +++ b/db/state/squeeze.go @@ -1020,10 +1020,8 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea } roTx.Rollback() - parallel := statecfg.ExperimentalParallelCommitment trieVariant := commitment.VariantHexPatriciaTrie - switch { - case parallel: + if statecfg.ExperimentalParallelCommitment { trieVariant = commitment.VariantParallelHexPatricia } @@ -1060,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 { + if statecfg.ExperimentalParallelCommitment { domains.EnableParaTrieDB(rwDb) } From 1aa696f5e439790d57fa10fe912f823b5c11cf4d Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 12 Aug 2026 11:06:38 +0700 Subject: [PATCH 3/4] execution/commitment: carry the streaming removal into the consolidated deepfold tests The five deepfold_*_regression_test.go files this branch edited were merged into deepfold_test.go by #23184 (a pure move), so the mode-table and helper changes had to be reapplied there. keyArena keeps main's remaining-based sizing: moving the type out of the deleted streaming_commitment.go dropped the field, but its call site still passes it. --- execution/commitment/deepfold_test.go | 38 ++++----------------- execution/commitment/streaming_deep_fold.go | 13 +++++-- 2 files changed, 17 insertions(+), 34 deletions(-) diff --git a/execution/commitment/deepfold_test.go b/execution/commitment/deepfold_test.go index 270b03f8d79..b63b7a6a819 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) diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go index 27e3e24cf07..af4c99c3726 100644 --- a/execution/commitment/streaming_deep_fold.go +++ b/execution/commitment/streaming_deep_fold.go @@ -312,15 +312,22 @@ 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. -type keyArena struct{ buf []byte } +// 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) { - a.buf = make([]byte, 0, max(keyArenaChunk, len(hk))) + 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)] From cc290e94b566f6598b2e3c7d413767361bcb9169 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 12 Aug 2026 11:59:55 +0700 Subject: [PATCH 4/4] execution/commitment: keep the keyArena tests with the type they cover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit keyArena outlived streaming_commitment.go — collectSubtreeKeys still uses it — but its two tests were in streaming_commitment_test.go and went with the file. They move to deepfold_test.go, next to the deep fold that drives them. --- execution/commitment/deepfold_test.go | 79 +++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/execution/commitment/deepfold_test.go b/execution/commitment/deepfold_test.go index b63b7a6a819..04397ffb42b 100644 --- a/execution/commitment/deepfold_test.go +++ b/execution/commitment/deepfold_test.go @@ -954,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)) + }) +}