execution/commitment: remove the streaming commitment mode

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)

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) {

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)
 

db/state/execctx/commitment_flag_test.go

@@ -33,13 +33,10 @@ import (
 
 func withCommitmentFlag(t *testing.T, variant commitment.TrieVariant) {
 	t.Helper()
-	origStream := statecfg.ExperimentalStreamingCommitment
 	origPar := statecfg.ExperimentalParallelCommitment
 	t.Cleanup(func() {
-		statecfg.ExperimentalStreamingCommitment = origStream
 		statecfg.ExperimentalParallelCommitment = origPar
 	})
-	statecfg.ExperimentalStreamingCommitment = variant == commitment.VariantStreamingHexPatricia
 	statecfg.ExperimentalParallelCommitment = variant == commitment.VariantParallelHexPatricia
 }
 
@@ -109,58 +106,3 @@ func TestSharedDomains_ParallelFlag_RootEquivalence(t *testing.T) {
 		"sequential and parallel commitment roots must match: sequential=%x parallel=%x",
 		seqRoot, parRoot)
 }
-
-func TestPickTrieVariant_StreamingFlag(t *testing.T) {
-	// No t.Parallel: mutates process-global statecfg flags.
-	withCommitmentFlag(t, commitment.VariantStreamingHexPatricia)
-	require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
-
-	statecfg.ExperimentalParallelCommitment = true
-	require.Equal(t, commitment.VariantStreamingHexPatricia, execctx.PickTrieVariant())
-
-	statecfg.ExperimentalStreamingCommitment = false
-	require.Equal(t, commitment.VariantParallelHexPatricia, execctx.PickTrieVariant())
-}
-
-func TestSharedDomains_StreamingFlag_RootEquivalence(t *testing.T) {
-	if testing.Short() {
-		t.Skip()
-	}
-	// No t.Parallel: mutates process-global statecfg flags.
-
-	stepSize := uint64(16)
-
-	runOnce := func(t *testing.T, streaming bool) []byte {
-		t.Helper()
-		variant := commitment.VariantHexPatriciaTrie
-		if streaming {
-			variant = commitment.VariantStreamingHexPatricia
-		}
-		withCommitmentFlag(t, variant)
-
-		db := newTestDb(t, stepSize)
-
-		ctx := t.Context()
-		rwTx, err := db.BeginTemporalRw(ctx)
-		require.NoError(t, err)
-		defer rwTx.Rollback()
-
-		sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
-		require.NoError(t, err)
-		defer sd.Close()
-
-		sd.EnableParaTrieDB(db)
-
-		got := sd.GetCommitmentCtx().Trie().Variant()
-		require.Equalf(t, variant, got, "trie variant for streaming=%v", streaming)
-
-		return runWriteCommitBatch(t, sd, rwTx)
-	}
-
-	seqRoot := runOnce(t, false)
-	strRoot := runOnce(t, true)
-
-	require.Equalf(t, seqRoot, strRoot,
-		"sequential and streaming commitment roots must match: sequential=%x streaming=%x",
-		seqRoot, strRoot)
-}

db/state/execctx/domain_shared.go

@@ -290,13 +290,7 @@ type SharedDomains struct {
 // entry points instead of leaving Variant unset and relying on an implicit
 // fallback inside the trie constructor.
 func PickTrieVariant() commitment.TrieVariant {
-	switch {
-	// Selecting more than one experimental-commitment flag is a misconfiguration;
-	// they are alternative paths. Streaming overlaps folding with execution, so it
-	// wins over parallel.
-	case statecfg.ExperimentalStreamingCommitment:
-		return commitment.VariantStreamingHexPatricia
-	case statecfg.ExperimentalParallelCommitment:
+	if statecfg.ExperimentalParallelCommitment {
 		return commitment.VariantParallelHexPatricia
 	}
 	return commitment.VariantHexPatriciaTrie

db/state/squeeze.go

@@ -1020,13 +1020,8 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
 		}
 		roTx.Rollback()
 
-		streaming := statecfg.ExperimentalStreamingCommitment
-		parallel := statecfg.ExperimentalParallelCommitment
 		trieVariant := commitment.VariantHexPatriciaTrie
-		switch {
-		case streaming:
-			trieVariant = commitment.VariantStreamingHexPatricia
-		case parallel:
+		if statecfg.ExperimentalParallelCommitment {
 			trieVariant = commitment.VariantParallelHexPatricia
 		}
 
@@ -1063,7 +1058,7 @@ func RebuildCommitmentFiles(ctx context.Context, rwDb kv.TemporalRwDB, txNumsRea
 			domains.SetTxNum(lastTxnumInShard - 1)
 			currentTxNum := lastTxnumInShard - 1
 			domains.GetCommitmentCtx().SetStateReader(commitmentdb.NewFilesOnlyStateReader(rwTx, lastTxnumInShard-1))
-			if parallel || streaming {
+			if statecfg.ExperimentalParallelCommitment {
 				domains.EnableParaTrieDB(rwDb)
 			}
 

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,

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")
-	})
 }

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:
 	}
 }

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 {