diff --git a/execution/builder/block_builder.go b/execution/builder/block_builder.go index e77019a08c2..5b396e61e00 100644 --- a/execution/builder/block_builder.go +++ b/execution/builder/block_builder.go @@ -28,19 +28,26 @@ import ( "github.com/erigontech/erigon/execution/types" ) -type BlockBuilderFunc func(param *Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) +// BlockBuilderFunc builds a payload. Its context ends when the payload is discarded, so anything +// that can block - opening a read view, waiting on a transaction provider - has to honour it. +type BlockBuilderFunc func(ctx context.Context, param *Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) -// BlockBuilder wraps a goroutine that builds Proof-of-Stake payloads (PoS "mining") +// BlockBuilder wraps a goroutine that builds Proof-of-Stake payloads (PoS "mining"). +// +// It answers to two different requests. Interrupting asks for the block it has so far, which is how +// a payload is collected. Discarding says the payload is not wanted at all, and cancels the work. type BlockBuilder struct { interrupt atomic.Bool + discard context.CancelFunc mu sync.Mutex done chan struct{} result *types.BlockWithReceipts err error } -func NewBlockBuilder(build BlockBuilderFunc, param *Parameters, maxBuildTime time.Duration) *BlockBuilder { - builder := &BlockBuilder{done: make(chan struct{})} +func NewBlockBuilder(ctx context.Context, build BlockBuilderFunc, param *Parameters, maxBuildTime time.Duration) *BlockBuilder { + buildCtx, discard := context.WithCancel(ctx) + builder := &BlockBuilder{done: make(chan struct{}), discard: discard} go func() { var result *types.BlockWithReceipts @@ -58,13 +65,18 @@ func NewBlockBuilder(build BlockBuilderFunc, param *Parameters, maxBuildTime tim builder.err = err builder.mu.Unlock() close(builder.done) + discard() }() log.Info("Building block...") t := time.Now() - result, err = build(param, &builder.interrupt) + result, err = build(buildCtx, param, &builder.interrupt) if err != nil { - log.Warn("Failed to build a block", "err", err) + if buildCtx.Err() != nil { + log.Debug("Block builder discarded", "err", err) + } else { + log.Warn("Failed to build a block", "err", err) + } } else { block := result.Block log.Info("Built block", "hash", block.Hash(), "height", block.NumberU64(), "txs", len(block.Transactions()), "executionRequests", len(result.Requests), "gasUsedPct", 100*float64(block.GasUsed())/float64(block.GasLimit()), "time", time.Since(t)) @@ -102,6 +114,28 @@ func (b *BlockBuilder) Stop(ctx context.Context) (*types.BlockWithReceipts, erro return b.result, b.err } +// Discard abandons the build and releases what it holds. A read view or a transaction provider +// blocked on the builder's context returns at once instead of waiting out its own deadline, which +// is the difference between an evicted builder freeing its resources now and freeing them a slot +// from now. +func (b *BlockBuilder) Discard() { + b.interrupt.Store(true) + b.discard() +} + +// Failed reports whether the builder has finished and ended in an error, which a caller looking to +// reuse it has to read as absent because that error is latched. +func (b *BlockBuilder) Failed() bool { + select { + case <-b.done: + default: + return false + } + b.mu.Lock() + defer b.mu.Unlock() + return b.err != nil +} + func (b *BlockBuilder) Block() *types.Block { b.mu.Lock() defer b.mu.Unlock() diff --git a/execution/builder/block_builder_test.go b/execution/builder/block_builder_test.go new file mode 100644 index 00000000000..5b2a1b8ca73 --- /dev/null +++ b/execution/builder/block_builder_test.go @@ -0,0 +1,84 @@ +// 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 builder + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/execution/types" +) + +func TestBlockBuilderRunningHasNotFailed(t *testing.T) { + t.Parallel() + + release := make(chan struct{}) + t.Cleanup(func() { close(release) }) + b := NewBlockBuilder(t.Context(), func(_ context.Context, _ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) { + <-release + return nil, errors.New("builder stopped") + }, &Parameters{}, time.Minute) + + require.Never(t, b.Failed, 50*time.Millisecond, 5*time.Millisecond) +} + +func TestBlockBuilderStoppedForItsPayloadHasNotFailed(t *testing.T) { + t.Parallel() + + b := NewBlockBuilder(t.Context(), func(_ context.Context, _ *Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil + }, &Parameters{}, time.Minute) + + _, err := b.Stop(t.Context()) + require.NoError(t, err) + + // Collecting the payload is what a proposal does. Reading that as failure would make a repeated + // request rebuild from scratch instead of being handed the block that was just built. + require.False(t, b.Failed()) +} + +func TestBlockBuilderHasFailedOnceItErrors(t *testing.T) { + t.Parallel() + + b := NewBlockBuilder(t.Context(), func(_ context.Context, _ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) { + return nil, errors.New("build failed") + }, &Parameters{}, time.Minute) + + require.Eventually(t, b.Failed, time.Second, time.Millisecond) +} + +func TestBlockBuilderStaysReusableOnceItFillsTheBlock(t *testing.T) { + t.Parallel() + + built := make(chan struct{}) + b := NewBlockBuilder(t.Context(), func(_ context.Context, _ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) { + defer close(built) + return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil + }, &Parameters{}, time.Minute) + + <-built + // A builder that ran out of room holds a complete payload, so its id is still worth reusing. + require.Never(t, b.Failed, 50*time.Millisecond, 5*time.Millisecond) +} diff --git a/execution/builder/builder.go b/execution/builder/builder.go index 5150348b674..95faef9a7b6 100644 --- a/execution/builder/builder.go +++ b/execution/builder/builder.go @@ -45,7 +45,6 @@ type SDProvider func() *execctx.SharedDomains // without staged-sync machinery. Its Build method satisfies BlockBuilderFunc and can // be passed directly to ExecModule. type Builder struct { - ctx context.Context db kv.TemporalRoDB pendingBlockCh chan *types.Block builderCfg *buildercfg.BuilderConfig @@ -64,7 +63,6 @@ type Builder struct { } func NewBuilder( - ctx context.Context, db kv.TemporalRoDB, builderCfg *buildercfg.BuilderConfig, chainConfig *chain.Config, @@ -81,7 +79,6 @@ func NewBuilder( logger log.Logger, ) *Builder { return &Builder{ - ctx: ctx, db: db, pendingBlockCh: make(chan *types.Block, 1), builderCfg: builderCfg, @@ -107,7 +104,10 @@ func (b *Builder) PendingBlockCh() chan *types.Block { } // Build satisfies BlockBuilderFunc. Pass b.Build directly to ExecModule. -func (b *Builder) Build(param *Parameters, interrupt *atomic.Bool) (result *types.BlockWithReceipts, err error) { +// +// Everything that can block runs under ctx, so discarding the payload releases the read view and +// unblocks the transaction provider instead of leaving them to finish on their own. +func (b *Builder) Build(ctx context.Context, param *Parameters, interrupt *atomic.Bool) (result *types.BlockWithReceipts, err error) { defer func() { if rec := recover(); rec != nil { err = fmt.Errorf("%+v, trace: %s", rec, dbg.Stack()) @@ -124,7 +124,7 @@ func (b *Builder) Build(param *Parameters, interrupt *atomic.Bool) (result *type BuiltBlock: &exec.AssembledBlock{}, } - tx, err := b.db.BeginTemporalRo(b.ctx) + tx, err := b.db.BeginTemporalRo(ctx) if err != nil { return nil, err } @@ -145,7 +145,7 @@ func (b *Builder) Build(param *Parameters, interrupt *atomic.Bool) (result *type } } - sd, err := execctx.NewSharedDomains(b.ctx, compositeTx, b.logger, execctx.WithoutDeferredBranchUpdates(), execctx.WithoutSharedBranchCache()) + sd, err := execctx.NewSharedDomains(ctx, compositeTx, b.logger, execctx.WithoutDeferredBranchUpdates(), execctx.WithoutSharedBranchCache()) if err != nil { return nil, err } @@ -172,10 +172,10 @@ func (b *Builder) Build(param *Parameters, interrupt *atomic.Bool) (result *type execCfg := StageBuilderExecCfg(state, b.notifier, b.chainConfig, b.engine, b.vmConfig, b.tmpdir, interrupt, param.PayloadId, txnProvider, b.blockReader) finishCfg := StageBuilderFinishCfg(b.chainConfig, b.engine, state, b.sealCancel, b.blockReader, b.latestBlockBuiltStore) - if err := createBlock(b.ctx, sd, compositeTx, executionAt, createCfg, b.logger); err != nil { + if err := createBlock(ctx, sd, compositeTx, executionAt, createCfg, b.logger); err != nil { return nil, err } - if err := execBlock(b.ctx, sd, compositeTx, executionAt, execCfg, b.executeBlockCfg, b.logger); err != nil { + if err := execBlock(ctx, sd, compositeTx, executionAt, execCfg, b.executeBlockCfg, b.logger); err != nil { return nil, err } if err := finishBlock(compositeTx, finishCfg, b.logger); err != nil { diff --git a/execution/builder/builder_test.go b/execution/builder/builder_test.go index f7c758522fe..87a74eb129a 100644 --- a/execution/builder/builder_test.go +++ b/execution/builder/builder_test.go @@ -48,14 +48,13 @@ func TestBuilder_Build_DBError(t *testing.T) { want := errors.New("db open failed") b := &Builder{ - ctx: context.Background(), db: &errDB{err: want}, builderCfg: &buildercfg.BuilderConfig{}, pendingBlockCh: make(chan *types.Block, 1), logger: log.New(), } - _, err := b.Build(&Parameters{}, &atomic.Bool{}) + _, err := b.Build(t.Context(), &Parameters{}, &atomic.Bool{}) require.ErrorIs(t, err, want) } diff --git a/execution/execmodule/block_building.go b/execution/execmodule/block_building.go index ef1d94cddaf..a8ce4392831 100644 --- a/execution/execmodule/block_building.go +++ b/execution/execmodule/block_building.go @@ -17,6 +17,7 @@ package execmodule import ( + "bytes" "context" "reflect" "time" @@ -60,16 +61,89 @@ func buildDuration(payloadTimestamp uint64, now time.Time, secondsPerSlot uint64 return min(max(d, slot/4), 2*slot) } +func cloneBuilderParameters(params *builder.Parameters) *builder.Parameters { + if params == nil { + return nil + } + cloned := *params + cloned.ExtraData = bytes.Clone(params.ExtraData) + if params.Withdrawals != nil { + cloned.Withdrawals = make([]*types.Withdrawal, len(params.Withdrawals)) + for i, withdrawal := range params.Withdrawals { + if withdrawal != nil { + copy := *withdrawal + cloned.Withdrawals[i] = © + } + } + } + if params.ParentBeaconBlockRoot != nil { + copy := *params.ParentBeaconBlockRoot + cloned.ParentBeaconBlockRoot = © + } + if params.SlotNumber != nil { + copy := *params.SlotNumber + cloned.SlotNumber = © + } + if params.TargetGasLimit != nil { + copy := *params.TargetGasLimit + cloned.TargetGasLimit = © + } + return &cloned +} + +// sameBuildRequest reports whether a request is asking for the payload another one is already +// building. A custom transaction provider is never treated as the same request: it is stateful and +// hands its transactions over once, so a second request carrying one is not asking for what the +// first is building, and comparing the provider itself would read fields the running build writes. +func sameBuildRequest(previous, current *builder.Parameters) bool { + if previous == nil || current == nil { + return false + } + if previous.CustomTxnProvider != nil || current.CustomTxnProvider != nil { + return false + } + return reflect.DeepEqual(previous, current) +} + +// builderEntry keeps a builder with the parameters and timestamp it was created for, so the +// three cannot drift apart and eviction can drop the timestamp index without scanning it. +type builderEntry struct { + builder *builder.BlockBuilder + params *builder.Parameters + timestamp uint64 +} + +func (e *ExecModule) dropBuilder(id uint64, entry *builderEntry) { + if e.buildersByTimestamp[entry.timestamp] == id { + delete(e.buildersByTimestamp, entry.timestamp) + } + delete(e.builders, id) +} + func (e *ExecModule) evictOldBuilders() { ids := common.SortedKeys(e.builders) // remove old builders so that at most MaxBuilders - 1 remain for i := 0; i <= len(e.builders)-engine_helpers.MaxBuilders; i++ { - delete(e.builders, ids[i]) + id := ids[i] + if old := e.builders[id]; old != nil { + if old.builder != nil { + old.builder.Discard() + } + if e.buildersByTimestamp[old.timestamp] == id { + delete(e.buildersByTimestamp, old.timestamp) + } + } + delete(e.builders, id) } } func (e *ExecModule) AssembleBlock(ctx context.Context, params *builder.Parameters) (AssembleBlockResult, error) { + // Cancellation is checked first so an expired request reports why it stopped instead of + // masquerading as contention, which callers retry. + if err := ctx.Err(); err != nil { + return AssembleBlockResult{}, err + } if !e.semaphore.TryAcquire(1) { return AssembleBlockResult{Busy: true}, nil } @@ -79,23 +153,35 @@ func (e *ExecModule) AssembleBlock(ctx context.Context, params *builder.Paramete return AssembleBlockResult{}, err } - // First check if we're already building a block with the requested parameters - if e.lastParameters != nil { - params.PayloadId = e.lastParameters.PayloadId - if reflect.DeepEqual(e.lastParameters, params) { - e.logger.Info("[ForkChoiceUpdated] duplicate build request") - return AssembleBlockResult{PayloadID: e.lastParameters.PayloadId}, nil + // A stopped builder is still worth reusing: it holds the payload it was stopped for, which is + // exactly what a repeated request is asking for. Only a failed one has to be passed over. + if previousID, ok := e.buildersByTimestamp[params.Timestamp]; ok { + if previous := e.builders[previousID]; previous != nil && previous.builder != nil && !previous.builder.Failed() { + params.PayloadId = previousID + if sameBuildRequest(previous.params, params) { + e.logger.Info("[ForkChoiceUpdated] duplicate build request") + return AssembleBlockResult{PayloadID: previousID}, nil + } } } - - // Initiate payload building + // A superseded builder keeps running to its own deadline. The timestamp index moves to the new + // one, so nothing reaches it by dedup, while an id already handed out goes on answering with a + // payload that is still growing. e.evictOldBuilders() e.nextPayloadId++ params.PayloadId = e.nextPayloadId - e.lastParameters = params + ownedParams := cloneBuilderParameters(params) - e.builders[e.nextPayloadId] = builder.NewBlockBuilder(e.builderFunc, params, buildDuration(params.Timestamp, time.Now(), e.config.SecondsPerSlot())) + if e.buildersByTimestamp == nil { + e.buildersByTimestamp = make(map[uint64]uint64) + } + e.builders[e.nextPayloadId] = &builderEntry{ + builder: builder.NewBlockBuilder(e.bacgroundCtx, e.builderFunc, ownedParams, buildDuration(params.Timestamp, time.Now(), e.config.SecondsPerSlot())), + params: ownedParams, + timestamp: params.Timestamp, + } + e.buildersByTimestamp[params.Timestamp] = e.nextPayloadId e.logger.Info("[ForkChoiceUpdated] BlockBuilder added", "payload", e.nextPayloadId) return AssembleBlockResult{PayloadID: e.nextPayloadId}, nil @@ -118,17 +204,29 @@ func blockValue(br *types.BlockWithReceipts, baseFee *uint256.Int) *uint256.Int } func (e *ExecModule) GetAssembledBlock(ctx context.Context, payloadID uint64) (AssembledBlockResult, error) { + if err := ctx.Err(); err != nil { + return AssembledBlockResult{}, err + } if !e.semaphore.TryAcquire(1) { return AssembledBlockResult{Busy: true}, nil } defer e.semaphore.Release(1) - bldr, ok := e.builders[payloadID] - if !ok { + entry, ok := e.builders[payloadID] + if !ok || entry == nil || entry.builder == nil { return AssembledBlockResult{}, nil } - blockWithReceipts, err := bldr.Stop(ctx) + blockWithReceipts, err := entry.builder.Stop(ctx) if err != nil { + // Stop reports the caller's wait expiring and the build's own failure through the same + // error, and a build can fail with a context error of its own - a transaction provider + // giving up, say. Only the builder knows which happened, so ask it rather than guess from + // the error: a caller that gave up leaves a builder still worth collecting. + if !entry.builder.Failed() { + return AssembledBlockResult{}, err + } + // Keeping a failed entry would hand its latched error to every retry. + e.dropBuilder(payloadID, entry) e.logger.Error("Failed to build PoS block", "err", err) return AssembledBlockResult{}, err } diff --git a/execution/execmodule/block_building_internal_test.go b/execution/execmodule/block_building_internal_test.go index b48479af2e6..1da8c27da6e 100644 --- a/execution/execmodule/block_building_internal_test.go +++ b/execution/execmodule/block_building_internal_test.go @@ -17,13 +17,517 @@ package execmodule import ( + "context" + "errors" + "fmt" "math" + "sync/atomic" "testing" "time" + "golang.org/x/sync/semaphore" + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common" + "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/execution/builder" + "github.com/erigontech/erigon/execution/chain" + "github.com/erigontech/erigon/execution/engineapi/engine_helpers" + "github.com/erigontech/erigon/execution/types" + "github.com/erigontech/erigon/txnprovider" ) +func TestAssembleBlockKeepsBuildersApartByTimestamp(t *testing.T) { + type runningBuilder struct { + id uint64 + interrupt *atomic.Bool + } + started := make(chan runningBuilder, 4) + module := &ExecModule{ + semaphore: semaphore.NewWeighted(1), + config: &chain.Config{}, + logger: log.Root(), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, params *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- runningBuilder{id: params.PayloadId, interrupt: interrupt} + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return nil, errors.New("builder stopped") + }, + } + t.Cleanup(func() { + for _, entry := range module.builders { + if entry != nil && entry.builder != nil { + _, _ = entry.builder.Stop(context.Background()) + } + } + }) + + waitStarted := func() runningBuilder { + t.Helper() + select { + case running := <-started: + return running + case <-time.After(time.Second): + t.Fatal("builder did not start") + return runningBuilder{} + } + } + assemble := func(timestamp uint64, parent common.Hash) (uint64, runningBuilder) { + t.Helper() + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: timestamp, ParentHash: parent}) + require.NoError(t, err) + require.False(t, result.Busy) + return result.PayloadID, waitStarted() + } + + firstID, first := assemble(100, common.Hash{0x01}) + adjacentID, adjacent := assemble(101, common.Hash{0x02}) + require.NotEqual(t, firstID, adjacentID) + + firstDuplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + require.Equal(t, firstID, firstDuplicate.PayloadID) + + // Superseding hands the timestamp to a new builder and leaves the old ones running, so only the + // index moves. Timestamp 101 is a different proposal and is untouched throughout. + secondID, second := assemble(100, common.Hash{0x03}) + require.NotEqual(t, firstID, secondID) + require.Equal(t, secondID, module.buildersByTimestamp[100]) + require.Equal(t, adjacentID, module.buildersByTimestamp[101]) + + thirdID, third := assemble(100, common.Hash{0x04}) + require.NotEqual(t, secondID, thirdID) + require.Equal(t, thirdID, module.buildersByTimestamp[100]) + + duplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x04}}) + require.NoError(t, err) + require.Equal(t, thirdID, duplicate.PayloadID) + + for _, running := range []runningBuilder{first, second, third, adjacent} { + require.False(t, running.interrupt.Load(), "builder %d must still be packing", running.id) + } + + // Eviction is where a builder is actually stopped, and it takes the timestamp index with it. + // Removing entries puts them beyond the registered cleanup, so release them here instead of + // leaving two goroutines running until their watchdogs fire. + module.builders[firstID].builder.Discard() + module.builders[secondID].builder.Discard() + delete(module.builders, firstID) + delete(module.builders, secondID) + for id := thirdID + 1; len(module.builders) < engine_helpers.MaxBuilders; id++ { + module.builders[id] = nil + } + module.evictOldBuilders() + require.Eventually(t, adjacent.interrupt.Load, time.Second, time.Millisecond) + require.NotContains(t, module.builders, adjacentID) + require.NotContains(t, module.buildersByTimestamp, uint64(101)) + require.False(t, third.interrupt.Load(), "the current builder for a timestamp must survive eviction") +} + +func TestEvictionReleasesABuilderBlockedOnItsProvider(t *testing.T) { + // A transaction provider can wait for most of a slot before returning, and the interrupt flag + // is not read until it does. Only cancelling the build reaches it. + entered := make(chan struct{}, 1) + released := make(chan error, 1) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(ctx context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + entered <- struct{}{} + select { + case <-ctx.Done(): + released <- ctx.Err() + return nil, ctx.Err() + case <-time.After(time.Minute): + released <- errors.New("provider was never released") + return nil, errors.New("provider was never released") + } + }, + } + + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + <-entered + + entry := module.builders[result.PayloadID] + require.NotNil(t, entry) + for id := result.PayloadID + 1; len(module.builders) < engine_helpers.MaxBuilders; id++ { + module.builders[id] = nil + } + module.evictOldBuilders() + require.NotContains(t, module.builders, result.PayloadID) + + // Observing the goroutine finish is the point: an evicted builder that merely has its flag set + // keeps its read view open until whatever it is blocked on gives up on its own. + select { + case err := <-released: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(5 * time.Second): + t.Fatal("evicted builder was never released") + } + require.Eventually(t, func() bool { return entry.builder.Failed() }, 5*time.Second, time.Millisecond) +} + +func TestSupersededBuilderKeepsPackingAndStaysRetrievable(t *testing.T) { + started := make(chan *atomic.Bool, 4) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- interrupt + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil + }, + } + + first, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + firstInterrupt := <-started + + second, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x02}}) + require.NoError(t, err) + require.NotEqual(t, first.PayloadID, second.PayloadID) + secondInterrupt := <-started + + // The timestamp index moves to the new builder, so nothing reaches the old one by dedup. It is + // left running: freezing it would answer an id already handed out with a near-empty payload. + require.Equal(t, second.PayloadID, module.buildersByTimestamp[100]) + require.False(t, firstInterrupt.Load()) + + assembled, err := module.GetAssembledBlock(t.Context(), first.PayloadID) + require.NoError(t, err) + require.NotNil(t, assembled.Block) + + secondInterrupt.Store(true) +} + +func TestCollectedPayloadIsHandedBackToARepeatedRequest(t *testing.T) { + started := make(chan struct{}, 4) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- struct{}{} + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil + }, + } + + params := func() *builder.Parameters { + return &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}} + } + first, err := module.AssembleBlock(t.Context(), params()) + require.NoError(t, err) + <-started + + assembled, err := module.GetAssembledBlock(t.Context(), first.PayloadID) + require.NoError(t, err) + require.NotNil(t, assembled.Block) + + // Collecting stops the builder. A repeated request must still be handed that payload: rebuilding + // from scratch this late means the next grab takes a near-empty block. + repeat, err := module.AssembleBlock(t.Context(), params()) + require.NoError(t, err) + require.Equal(t, first.PayloadID, repeat.PayloadID) + require.Empty(t, started, "a repeated request must not start a second builder") +} + +func TestAssembleBlockDoesNotReuseFailedBuilder(t *testing.T) { + var failNext atomic.Bool + failNext.Store(true) + started := make(chan struct{}, 4) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- struct{}{} + if failNext.Swap(false) { + return nil, errors.New("build failed") + } + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return nil, errors.New("builder stopped") + }, + } + + params := func() *builder.Parameters { + return &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}} + } + first, err := module.AssembleBlock(t.Context(), params()) + require.NoError(t, err) + <-started + require.Eventually(t, module.builders[first.PayloadID].builder.Failed, time.Second, time.Millisecond) + + // Identical parameters would normally dedup onto the same id. A builder that already died + // latches its error, so reusing it would spend the slot on a payload that can never arrive. + second, err := module.AssembleBlock(t.Context(), params()) + require.NoError(t, err) + require.NotEqual(t, first.PayloadID, second.PayloadID) + <-started + + _, _ = module.builders[second.PayloadID].builder.Stop(context.Background()) +} + +func TestGetAssembledBlockDropsFailedBuilder(t *testing.T) { + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) { + return nil, errors.New("build failed") + }, + } + + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + + _, err = module.GetAssembledBlock(t.Context(), result.PayloadID) + require.Error(t, err) + + // The error is latched, so leaving the entry in place would keep serving it to every retry. + require.NotContains(t, module.builders, result.PayloadID) + require.NotContains(t, module.buildersByTimestamp, uint64(100)) +} + +func TestGetAssembledBlockKeepsBuilderWhenTheCallerGivesUpMidStop(t *testing.T) { + interrupted := make(chan struct{}) + release := make(chan struct{}) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + close(interrupted) + <-release + return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil + }, + } + + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + + // Cancel while Stop is already waiting, which is the window a caller-side timeout actually + // lands in. Cancelling beforehand returns at the entry check and exercises none of this. + ctx, cancel := context.WithCancel(t.Context()) + collected := make(chan error, 1) + go func() { + _, collectErr := module.GetAssembledBlock(ctx, result.PayloadID) + collected <- collectErr + }() + <-interrupted + cancel() + require.ErrorIs(t, <-collected, context.Canceled) + + // The builder was not dropped, so the payload it goes on to produce is still reachable. + require.Contains(t, module.builders, result.PayloadID) + require.Equal(t, result.PayloadID, module.buildersByTimestamp[100]) + + close(release) + assembled, err := module.GetAssembledBlock(t.Context(), result.PayloadID) + require.NoError(t, err) + require.NotNil(t, assembled.Block) +} + +// mutableTxnProvider stands in for the stateful providers the testing namespace supplies: it hands +// its transactions over once and clears them, from the build goroutine. +type mutableTxnProvider struct { + txns []types.Transaction + done atomic.Bool +} + +func (m *mutableTxnProvider) ProvideTxns(context.Context, ...txnprovider.ProvideOption) ([]types.Transaction, error) { + if !m.done.CompareAndSwap(false, true) { + return nil, nil + } + txns := m.txns + m.txns = nil + return txns, nil +} + +func TestAssembleBlockNeverReusesABuilderWithACustomProvider(t *testing.T) { + started := make(chan struct{}, 4) + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(ctx context.Context, params *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- struct{}{} + // Keep the provider busy for the whole test, which is when a comparison would read it. + for !interrupt.Load() && ctx.Err() == nil { + if params.CustomTxnProvider != nil { + _, _ = params.CustomTxnProvider.ProvideTxns(ctx) + } + time.Sleep(time.Millisecond) + } + return nil, errors.New("builder stopped") + }, + } + + withProvider := func() *builder.Parameters { + return &builder.Parameters{ + Timestamp: 100, + ParentHash: common.Hash{0x01}, + CustomTxnProvider: &mutableTxnProvider{txns: []types.Transaction{}}, + } + } + first, err := module.AssembleBlock(t.Context(), withProvider()) + require.NoError(t, err) + <-started + + // The provider is single-shot and mutates as it runs, so a second request carrying one is not + // asking for what the first is building, and its fields must never be compared. + second, err := module.AssembleBlock(t.Context(), withProvider()) + require.NoError(t, err) + require.NotEqual(t, first.PayloadID, second.PayloadID) + <-started + + for _, entry := range module.builders { + entry.builder.Discard() + } +} + +func TestAssembleBlockOwnsParameters(t *testing.T) { + type observedParameters struct { + parentRoot common.Hash + extraData byte + } + readParameters := make(chan struct{}) + observed := make(chan observedParameters, 1) + module := &ExecModule{ + semaphore: semaphore.NewWeighted(1), + config: &chain.Config{}, + logger: log.Root(), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, params *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + <-readParameters + observed <- observedParameters{parentRoot: *params.ParentBeaconBlockRoot, extraData: params.ExtraData[0]} + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return nil, errors.New("builder stopped") + }, + } + root := common.Hash{0xaa} + params := &builder.Parameters{ + Timestamp: 100, + ParentHash: common.Hash{0x01}, + ParentBeaconBlockRoot: &root, + ExtraData: []byte{0xbb}, + } + result, err := module.AssembleBlock(t.Context(), params) + require.NoError(t, err) + require.False(t, result.Busy) + + root[0] = 0xcc + params.ExtraData[0] = 0xdd + close(readParameters) + require.Equal(t, observedParameters{parentRoot: common.Hash{0xaa}, extraData: 0xbb}, <-observed) + + duplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{ + Timestamp: 100, + ParentHash: common.Hash{0x01}, + ParentBeaconBlockRoot: &common.Hash{0xaa}, + ExtraData: []byte{0xbb}, + }) + require.NoError(t, err) + require.Equal(t, result.PayloadID, duplicate.PayloadID) + _, _ = module.builders[result.PayloadID].builder.Stop(context.Background()) +} + +func TestAssembleBlockCanceledContextDoesNotSupersedeBuilder(t *testing.T) { + started := make(chan *atomic.Bool, 1) + module := &ExecModule{ + semaphore: semaphore.NewWeighted(1), + config: &chain.Config{}, + logger: log.Root(), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(_ context.Context, _ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) { + started <- interrupt + for !interrupt.Load() { + time.Sleep(time.Millisecond) + } + return nil, errors.New("builder stopped") + }, + } + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + interrupt := <-started + t.Cleanup(func() { + _, _ = module.builders[result.PayloadID].builder.Stop(context.Background()) + }) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = module.AssembleBlock(ctx, &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x02}}) + require.ErrorIs(t, err, context.Canceled) + require.False(t, interrupt.Load()) + require.Equal(t, result.PayloadID, module.buildersByTimestamp[100]) +} + +func TestCloneBuilderParametersPreservesRepresentations(t *testing.T) { + require.Nil(t, cloneBuilderParameters(nil)) + + empty := cloneBuilderParameters(&builder.Parameters{Withdrawals: []*types.Withdrawal{}, ExtraData: []byte{}}) + require.NotNil(t, empty.Withdrawals) + require.NotNil(t, empty.ExtraData) + + root := common.Hash{0x01} + slot := uint64(2) + gasLimit := uint64(3) + params := &builder.Parameters{ + Withdrawals: []*types.Withdrawal{nil, {Index: 4}}, + ParentBeaconBlockRoot: &root, + SlotNumber: &slot, + TargetGasLimit: &gasLimit, + ExtraData: []byte{5}, + } + cloned := cloneBuilderParameters(params) + params.Withdrawals[1].Index = 40 + root[0] = 10 + slot = 20 + gasLimit = 30 + params.ExtraData[0] = 50 + + require.Nil(t, cloned.Withdrawals[0]) + require.Equal(t, uint64(4), cloned.Withdrawals[1].Index) + require.Equal(t, common.Hash{0x01}, *cloned.ParentBeaconBlockRoot) + require.Equal(t, uint64(2), *cloned.SlotNumber) + require.Equal(t, uint64(3), *cloned.TargetGasLimit) + require.Equal(t, byte(5), cloned.ExtraData[0]) +} + func TestBuildDuration(t *testing.T) { const ethereum, gnosis = uint64(12), uint64(5) slotStart := time.Unix(1_700_000_000, 0) @@ -70,3 +574,28 @@ func TestBuildDurationCapsOverflowingTimestamp(t *testing.T) { // the floor instead of the cap. require.Equal(t, 24*time.Second, buildDuration(math.MaxUint64, now, 12)) } + +func TestGetAssembledBlockDropsABuildThatFailedWithAContextError(t *testing.T) { + module := &ExecModule{ + logger: log.Root(), + config: &chain.Config{}, + semaphore: semaphore.NewWeighted(1), + builders: map[uint64]*builderEntry{}, + bacgroundCtx: t.Context(), + builderFunc: func(context.Context, *builder.Parameters, *atomic.Bool) (*types.BlockWithReceipts, error) { + // A transaction provider that gives up reports its own context error, which is a failed + // build rather than a caller that stopped waiting. + return nil, fmt.Errorf("issue while waiting for parent block: %w", context.DeadlineExceeded) + }, + } + + result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}) + require.NoError(t, err) + + _, err = module.GetAssembledBlock(t.Context(), result.PayloadID) + require.ErrorIs(t, err, context.DeadlineExceeded) + + // Keeping it would serve that latched error to every later retry of the same slot. + require.NotContains(t, module.builders, result.PayloadID) + require.NotContains(t, module.buildersByTimestamp, uint64(100)) +} diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go index 6d771fa4001..292b7b70df7 100644 --- a/execution/execmodule/exec_module.go +++ b/execution/execmodule/exec_module.go @@ -194,10 +194,10 @@ type ExecModule struct { logger log.Logger // Block building - nextPayloadId uint64 - lastParameters *builder.Parameters - builderFunc builder.BlockBuilderFunc - builders map[uint64]*builder.BlockBuilder + nextPayloadId uint64 + builderFunc builder.BlockBuilderFunc + builders map[uint64]*builderEntry + buildersByTimestamp map[uint64]uint64 // Changes accumulator hook *stageloop.Hook @@ -266,7 +266,7 @@ func NewExecModule( logger: logger, forkValidator: forkValidator, pipelineExecutor: pipelineExecutor, - builders: make(map[uint64]*builder.BlockBuilder), + builders: make(map[uint64]*builderEntry), builderFunc: builderFunc, config: config, semaphore: semaphore.NewWeighted(1), diff --git a/execution/execmodule/execmoduletester/exec_module_tester.go b/execution/execmodule/execmoduletester/exec_module_tester.go index bba37c9692f..17f361de158 100644 --- a/execution/execmodule/execmoduletester/exec_module_tester.go +++ b/execution/execmodule/execmoduletester/exec_module_tester.go @@ -681,7 +681,6 @@ func New(tb testing.TB, opts ...Option) *ExecModuleTester { readAheader := exec.NewBlockReadAheader() blkBuilder := builder.NewBuilder( - mock.Ctx, mock.DB, &cfg.Builder, mock.ChainConfig, diff --git a/node/eth/backend.go b/node/eth/backend.go index 7ab53c07317..c3fb1f71e61 100644 --- a/node/eth/backend.go +++ b/node/eth/backend.go @@ -806,7 +806,6 @@ func New(ctx context.Context, stack *node.Node, config *ethconfig.Config, logger } blkBuilder := builder.NewBuilder( - backend.sentryCtx, backend.chainDB, &config.Builder, backend.chainConfig,