From 8208e2156eb3f90bddd36d471abeec023bcc8905 Mon Sep 17 00:00:00 2001 From: pk910 Date: Mon, 7 Sep 2026 11:48:22 +0200 Subject: [PATCH 1/4] fix(payload_builder): supersede synthesized-attribute builds with node attributes The missing-block fallback fires at the slot's build start time and copies the previous slot's attributes. Teku emits the next slot's attributes only ~30 ms before that point, so a delayed emission (the epoch transition on glamsterdam-devnet-8 slot 172896 cost ~0.8 s) lets the fallback win the race: the build ran from pre-transition withdrawal amounts, the bid was accepted (the randao mix of an epoch's first slot equals the previous one) and every node rejected the envelope with a withdrawals mismatch. The real attributes arrived 0.8 s later and were ignored because a build for the same parent tuple had already started. Fallback events now carry a Synthesized marker and every started build keeps its cancel func. A node-received event for the same parent tuple supersedes a build that ran from synthesized attributes unless the build inputs are identical: the in-flight engine build is cancelled, an already emitted payload is withdrawn from the cache (recorded as a failed build, reason "superseded by beacon node attributes") and the tuple is released so the late-build path rebuilds from the node's attributes. Publishing a payload and superseding it serialize on the build lock, so a stale payload can never be emitted after its abort. --- CLAUDE.md | 14 + pkg/payload_builder/build_supersede_test.go | 300 ++++++++++++++++++++ pkg/payload_builder/payload_cache.go | 24 ++ pkg/payload_builder/service.go | 194 +++++++++++-- pkg/rpc/beacon/events.go | 45 +++ pkg/rpc/beacon/events_test.go | 84 ++++++ pkg/slot_results/tracker.go | 7 +- 7 files changed, 645 insertions(+), 23 deletions(-) create mode 100644 pkg/payload_builder/build_supersede_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 84c3e9e..f876470 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -244,6 +244,20 @@ npm run clean - Calls Engine API to construct execution payloads (forkchoiceUpdated → getPayload) - Emits `PayloadReadyEvent` to subscribers; plan-involved skips fire `BuildSkippedEvent` (deduped per slot) for the slot results tracker + - **Missing-block fallback** (`applyAttributesFallback`): when no attributes + arrived for a slot by its build start time, the previous slot's are + re-used with the proposal slot advanced and the event marked + `Synthesized`. Such attributes are stale whenever the state moved in + between (an epoch transition changes the expected withdrawals), and Teku + emits next-slot attributes only ~30 ms before the build start, so the + fallback can win the race against a late node event. A node-received + event for the same parent tuple therefore SUPERSEDES a build that ran + from synthesized attributes (`supersedeSynthesizedBuild`): unless its + build inputs are identical (`PayloadAttributesEvent.BuildInputsEqual`), + the in-flight engine build is cancelled, an already emitted payload is + withdrawn from the payload cache (a `PayloadBuildFailedEvent` with reason + `superseded by beacon node attributes` is recorded) and the tuple is + rebuilt from the node's attributes via the late-build path 2. **Payload Bidder** (`pkg/payload_bidder/`) — shared Gloas+ bid/reveal domain - `Signer`, `BuildSignedBid`, `BuildSignedEnvelope`: bid/envelope construction + signing diff --git a/pkg/payload_builder/build_supersede_test.go b/pkg/payload_builder/build_supersede_test.go new file mode 100644 index 0000000..de1c7d7 --- /dev/null +++ b/pkg/payload_builder/build_supersede_test.go @@ -0,0 +1,300 @@ +package payload_builder + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethpandaops/go-eth2-client/spec/capella" + "github.com/ethpandaops/go-eth2-client/spec/phase0" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/buildoor/pkg/action_plan" + "github.com/ethpandaops/buildoor/pkg/chain" + "github.com/ethpandaops/buildoor/pkg/config" + "github.com/ethpandaops/buildoor/pkg/rpc/beacon" +) + +// pastChainService places every slot just in the past: its build start time +// has passed (late attributes take the late-build path) while the slot has +// not ended yet (late builds are still accepted). +type pastChainService struct { + *stubChainService + + genesis time.Time +} + +// newPastChainService anchors genesis so that testSlot started one second ago. +func newPastChainService(spec *chain.ChainSpec, testSlot phase0.Slot) *pastChainService { + slotDuration := time.Duration(testSlot) * spec.SecondsPerSlot + + return &pastChainService{ + stubChainService: &stubChainService{spec: spec}, + genesis: time.Now().Add(-time.Second - slotDuration), + } +} + +func (s *pastChainService) SlotToTime(slot phase0.Slot) time.Time { + return s.genesis.Add(time.Duration(slot) * s.spec.SecondsPerSlot) +} + +func supersedeTestService(t *testing.T, chainSvc chain.Service) *Service { + t.Helper() + + log := logrus.New() + log.SetLevel(logrus.PanicLevel) + + cfg := config.DefaultConfig() + cfg.EPBSEnabled = true + planSvc := action_plan.NewPlanService(cfg, chainSvc, log) + + clClient, err := beacon.NewClient(context.Background(), "http://127.0.0.1:1", log) + require.NoError(t, err) + + svc, err := NewService(cfg, clClient, chainSvc, planSvc, nil, common.Address{}, log) + require.NoError(t, err) + + svc.ctx = context.Background() + svc.payloadBuilder = NewPayloadBuilder(clClient, nil, chainSvc, common.Address{}, cfg, log, nil) + + return svc +} + +// synthesizedAttrs returns the attributes the missing-block fallback would +// synthesize for slot 133 from slot 132 (pre-epoch-transition withdrawals). +func synthesizedAttrs() *beacon.PayloadAttributesEvent { + return &beacon.PayloadAttributesEvent{ + ProposalSlot: 133, + ProposerIndex: 4, + ParentBlockRoot: phase0.Root{0x01}, + ParentBlockHash: phase0.Hash32{0xaa}, + Timestamp: 1396, + PrevRandao: phase0.Root{0xcc}, + Withdrawals: []*capella.Withdrawal{ + {Index: 1, ValidatorIndex: 10, Amount: 84045}, + }, + Synthesized: true, + } +} + +// registerBuild registers a started build for the event's tuple the way +// executeCandidateBuild does, returning the build and the context it would +// run the engine build with. +func registerBuild( + svc *Service, attrs *beacon.PayloadAttributesEvent, synthesized bool, +) (*candidateBuild, context.Context) { + ctx, cancel := context.WithCancel(context.Background()) + + svc.scheduledBuildMu.Lock() + defer svc.scheduledBuildMu.Unlock() + + state := svc.slotBuilds[attrs.ProposalSlot] + if state == nil { + state = newSlotBuildState() + state.passScheduled = true + svc.slotBuilds[attrs.ProposalSlot] = state + } + + build := &candidateBuild{attrs: attrs, synthesized: synthesized, cancel: cancel} + state.started[beacon.AttrParentKeyOf(attrs)] = build + + return build, ctx +} + +func (s *Service) startedBuild(slot phase0.Slot, key beacon.AttrParentKey) *candidateBuild { + s.scheduledBuildMu.Lock() + defer s.scheduledBuildMu.Unlock() + + if state := s.slotBuilds[slot]; state != nil { + return state.started[key] + } + + return nil +} + +func TestSupersedeSynthesizedBuild_AbortsInFlightBuild(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + svc := supersedeTestService(t, &stubChainService{spec: spec}) + + synthesized := synthesizedAttrs() + build, ctx := registerBuild(svc, synthesized, true) + + failedSub := svc.SubscribePayloadBuildFailed(4, false) + defer failedSub.Unsubscribe() + + // The node's attributes carry the post-transition withdrawal amounts. + real := *synthesized + real.Synthesized = false + real.Withdrawals = []*capella.Withdrawal{{Index: 1, ValidatorIndex: 10, Amount: 104426}} + + require.True(t, svc.supersedeSynthesizedBuild(&real)) + + assert.Error(t, ctx.Err(), "the in-flight engine build is cancelled") + assert.True(t, svc.buildAborted(build)) + assert.Nil(t, svc.startedBuild(133, beacon.AttrParentKeyOf(&real)), + "tuple released for the rebuild") + + select { + case <-failedSub.Channel(): + t.Fatal("a running build reports its own abort, supersede must not report it") + default: + } + + // The build goroutine finishing after the abort never emits its payload. + require.False(t, svc.supersedeSynthesizedBuild(&real), "nothing left to supersede") +} + +func TestSupersedeSynthesizedBuild_KeepsEqualInputs(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + svc := supersedeTestService(t, &stubChainService{spec: spec}) + + synthesized := synthesizedAttrs() + build, ctx := registerBuild(svc, synthesized, true) + + real := *synthesized + real.Synthesized = false + real.ParentBlockNumber = 31 // informational, backfilled by sanitization + real.Withdrawals = []*capella.Withdrawal{{Index: 1, ValidatorIndex: 10, Amount: 84045}} + + require.False(t, svc.supersedeSynthesizedBuild(&real), "identical inputs keep the build") + assert.NoError(t, ctx.Err()) + assert.False(t, build.synthesized, "the build now counts as confirmed by the node") + assert.Same(t, build, svc.startedBuild(133, beacon.AttrParentKeyOf(&real))) + + // Once confirmed, a later differing node event no longer aborts it (the + // regular variant handling applies). + differing := real + differing.Timestamp++ + require.False(t, svc.supersedeSynthesizedBuild(&differing)) + assert.NoError(t, ctx.Err()) +} + +func TestSupersedeSynthesizedBuild_IgnoresNodeBuilds(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + svc := supersedeTestService(t, &stubChainService{spec: spec}) + + attrs := synthesizedAttrs() + attrs.Synthesized = false + _, ctx := registerBuild(svc, attrs, false) + + updated := *attrs + updated.Timestamp++ + + require.False(t, svc.supersedeSynthesizedBuild(&updated)) + assert.NoError(t, ctx.Err()) + + // Unknown slot / tuple: nothing to do. + other := *attrs + other.ProposalSlot = 99 + require.False(t, svc.supersedeSynthesizedBuild(&other)) +} + +func TestSupersedeSynthesizedBuild_WithdrawsEmittedPayload(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + svc := supersedeTestService(t, &stubChainService{spec: spec}) + + synthesized := synthesizedAttrs() + build, _ := registerBuild(svc, synthesized, true) + + // The build already finished: its payload is cached and was dispatched. + payload := &Payload{ + Attributes: synthesized, + BlockHash: phase0.Hash32{0xb0, 0xd7}, + Candidate: chain.CandidateParentFull, + ReadyAt: time.Now(), + } + + svc.scheduledBuildMu.Lock() + build.payload = payload + svc.payloadCache.Store(payload) + svc.scheduledBuildMu.Unlock() + + failedSub := svc.SubscribePayloadBuildFailed(4, false) + defer failedSub.Unsubscribe() + + real := *synthesized + real.Synthesized = false + real.Withdrawals = []*capella.Withdrawal{{Index: 1, ValidatorIndex: 10, Amount: 104426}} + + require.True(t, svc.supersedeSynthesizedBuild(&real)) + + assert.Nil(t, svc.payloadCache.Get(133), "the stale payload must never be bid") + assert.Nil(t, svc.payloadCache.GetByBlockHash(payload.BlockHash)) + + select { + case event := <-failedSub.Channel(): + assert.Equal(t, phase0.Slot(133), event.Slot) + assert.Equal(t, string(chain.CandidateParentFull), event.Candidate) + assert.Equal(t, errBuildSuperseded, event.Error) + default: + t.Fatal("a finished build must be reported as superseded") + } +} + +// TestHandlePayloadAttributes_RebuildsAfterSupersede drives the whole path +// through the attributes handler: the fallback's synthesized build is +// aborted by the node's differing event and a fresh build starts from the +// node's attributes on the same parent tuple. +func TestHandlePayloadAttributes_RebuildsAfterSupersede(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + chainSvc := newPastChainService(spec, 133) + svc := supersedeTestService(t, chainSvc) + + synthesized := synthesizedAttrs() + require.True(t, svc.clClient.Events().InjectPayloadAttributes(synthesized)) + + stale, ctx := registerBuild(svc, synthesized, true) + key := beacon.AttrParentKeyOf(synthesized) + + failedSub := svc.SubscribePayloadBuildFailed(4, false) + defer failedSub.Unsubscribe() + + real := *synthesized + real.Synthesized = false + real.Withdrawals = []*capella.Withdrawal{{Index: 1, ValidatorIndex: 10, Amount: 104426}} + + svc.handlePayloadAttributesEvent(&real) + + assert.Error(t, ctx.Err(), "synthesized build aborted") + assert.True(t, svc.buildAborted(stale)) + + // The late-build path starts a fresh build from the node's attributes + // (it fails against the offline clients, which is fine: it ran). + require.Eventually(t, func() bool { + build := svc.startedBuild(133, key) + return build != nil && build != stale && !build.synthesized + }, 2*time.Second, 10*time.Millisecond, "a new build from the node's attributes must start") + + select { + case event := <-failedSub.Channel(): + assert.NotEqual(t, errBuildSuperseded, event.Error, "the rebuild fails for its own reason") + case <-time.After(2 * time.Second): + t.Fatal("the rebuild must report its outcome") + } + + // A synthesized event never supersedes anything. + again := *synthesized + svc.handlePayloadAttributesEvent(&again) + assert.NotNil(t, svc.startedBuild(133, key)) +} + +func TestPayloadCacheRemove(t *testing.T) { + cache := NewPayloadCache(4) + attrs := synthesizedAttrs() + + first := &Payload{Attributes: attrs, BlockHash: phase0.Hash32{0x01}} + cache.Store(first) + + replacement := &Payload{Attributes: attrs, BlockHash: phase0.Hash32{0x02}} + cache.Store(replacement) + + assert.False(t, cache.Remove(first), "a newer build on the tuple stays") + assert.Same(t, replacement, cache.Get(133)) + + assert.True(t, cache.Remove(replacement)) + assert.Nil(t, cache.Get(133)) + assert.Equal(t, 0, cache.Size()) +} diff --git a/pkg/payload_builder/payload_cache.go b/pkg/payload_builder/payload_cache.go index 6bfc1b2..377e6db 100644 --- a/pkg/payload_builder/payload_cache.go +++ b/pkg/payload_builder/payload_cache.go @@ -150,6 +150,30 @@ func (c *PayloadCache) GetByBlockHash(blockHash phase0.Hash32) *Payload { return nil } +// Remove drops the given payload from the cache when it is still the cached +// build of its parent tuple (a newer build on the same tuple stays). Returns +// whether the payload was removed. +func (c *PayloadCache) Remove(payload *Payload) bool { + c.mu.Lock() + defer c.mu.Unlock() + + slot := payload.Attributes.ProposalSlot + key := beacon.AttrParentKeyOf(payload.Attributes) + + variants := c.payloads[slot] + if variants[key] != payload { + return false + } + + delete(variants, key) + + if len(variants) == 0 { + delete(c.payloads, slot) + } + + return true +} + // Delete removes all payloads for the given slot. func (c *PayloadCache) Delete(slot phase0.Slot) { c.mu.Lock() diff --git a/pkg/payload_builder/service.go b/pkg/payload_builder/service.go index 51b8348..b9aee41 100644 --- a/pkg/payload_builder/service.go +++ b/pkg/payload_builder/service.go @@ -414,6 +414,13 @@ func (s *Service) handlePayloadAttributesEvent(event *beacon.PayloadAttributesEv s.markPayloadWon(event.ParentBlockHash, payload.Attributes.ProposalSlot) } + // The beacon node's own attributes supersede a build that started from + // synthesized ones for the same parent (the fallback won the race against + // a late node event); the tuple is released so the paths below rebuild. + if !event.Synthesized { + s.supersedeSynthesizedBuild(event) + } + // Arm the missing-block fallback for the NEXT proposal slot: if its // block goes missing entirely, some clients never emit fresh attributes // and this slot's attributes get re-used instead. @@ -472,14 +479,31 @@ func (s *Service) handlePayloadAttributesEvent(event *beacon.PayloadAttributesEv type slotBuildState struct { passScheduled bool buildStartMs int64 - started map[beacon.AttrParentKey]bool + started map[beacon.AttrParentKey]*candidateBuild readyFired bool // OnSlotBuilt/stat accounting fired (once per slot) } func newSlotBuildState() *slotBuildState { - return &slotBuildState{started: make(map[beacon.AttrParentKey]bool, 4)} + return &slotBuildState{started: make(map[beacon.AttrParentKey]*candidateBuild, 4)} } +// candidateBuild is one started candidate build of a slot, tracked by the +// parent tuple it builds on. A build from locally synthesized attributes +// (missing-block fallback or candidate synthesis) can be superseded by the +// beacon node's own attributes for the same tuple while it runs or after it +// emitted its payload; the fields are guarded by Service.scheduledBuildMu. +type candidateBuild struct { + attrs *beacon.PayloadAttributesEvent // attributes the build ran from + synthesized bool // attrs were synthesized locally + cancel context.CancelFunc // aborts the in-flight engine build + aborted bool // superseded: the payload must not be emitted + payload *Payload // emitted payload (nil while building or failed) +} + +// errBuildSuperseded is the recorded failure reason of a build superseded by +// the beacon node's attributes. +const errBuildSuperseded = "superseded by beacon node attributes" + // maybeLateBuild activates a candidate build for an attributes variant that // arrived after the slot's build pass already ran: the chain moved (reorg, // payload-miss flip, late reveal) and the new parent still deserves a payload @@ -489,7 +513,7 @@ func (s *Service) maybeLateBuild(slot phase0.Slot, event *beacon.PayloadAttribut s.scheduledBuildMu.Lock() state := s.slotBuilds[slot] - alreadyStarted := state != nil && state.started[beacon.AttrParentKeyOf(event)] + alreadyStarted := state != nil && state.started[beacon.AttrParentKeyOf(event)] != nil s.scheduledBuildMu.Unlock() if alreadyStarted || time.Now().After(s.slotEndTime(slot)) { @@ -635,6 +659,7 @@ func (s *Service) applyAttributesFallback(targetSlot phase0.Slot) { synthesized := *parent synthesized.ProposalSlot = targetSlot + synthesized.Synthesized = true synthesized.Timestamp = parent.Timestamp + skippedSlots*uint64(s.chainSvc.GetChainSpec().SecondsPerSlot.Seconds()) @@ -647,14 +672,18 @@ func (s *Service) applyAttributesFallback(targetSlot phase0.Slot) { "Cannot resolve proposer for synthesized attributes, keeping the source slot's") } - // The randao mix rotates at epoch boundaries; a value copied across one - // is invalid and any payload built from it will be rejected. + // The epoch transition changes the state the attributes derive from: the + // expected withdrawals move with the rewards applied to the swept + // validators and the randao mix rotates. Copied across an epoch boundary + // they are stale and a payload built from them is rejected; the build + // stays only until the beacon node's own attributes supersede it. spec := s.chainSvc.GetChainSpec() if uint64(targetSlot)/spec.SlotsPerEpoch != uint64(parent.ProposalSlot)/spec.SlotsPerEpoch { s.log.WithFields(logrus.Fields{ "slot": targetSlot, "attrs_from": parent.ProposalSlot, - }).Warn("Synthesized attributes cross an epoch boundary, prev_randao may be stale") + }).Warn("Synthesized attributes cross an epoch boundary, " + + "withdrawals and prev_randao may be stale") } if !events.InjectPayloadAttributes(&synthesized) { @@ -771,12 +800,28 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { } tuple := beacon.AttrParentKeyOf(target.attrs) - if state.started[tuple] { + if state.started[tuple] != nil { s.scheduledBuildMu.Unlock() return } - state.started[tuple] = true + // Size the build deadline to the target's build time plus a margin for the + // engine getPayload and finality lookups, so a long PayloadBuildTime doesn't + // make the getPayload call time out spuriously. The cancel func is + // registered with the build so a superseding attributes event can abort + // the engine build while it runs. + buildTimeMs := s.candidateBuildTime(target) + buildTimeout := time.Duration(buildTimeMs)*time.Millisecond + buildCallTimeout + ctx, cancel := context.WithTimeout(s.ctx, buildTimeout) + + defer cancel() + + build := &candidateBuild{ + attrs: target.attrs, + synthesized: target.derived || target.attrs.Synthesized, + cancel: cancel, + } + state.started[tuple] = build s.scheduledBuildMu.Unlock() // The frozen plan (idempotent Freeze) decides whether to build this slot's @@ -801,17 +846,13 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { StartedAt: time.Now(), }) - // Size the build deadline to the target's build time plus a margin for the - // engine getPayload and finality lookups, so a long PayloadBuildTime doesn't - // make the getPayload call time out spuriously. - buildTimeMs := s.candidateBuildTime(target) - buildTimeout := time.Duration(buildTimeMs)*time.Millisecond + buildCallTimeout - ctx, cancel := context.WithTimeout(s.ctx, buildTimeout) - - defer cancel() - payloadEvent, err := s.payloadBuilder.BuildPayloadFromAttributes(ctx, event, buildTimeMs) if err != nil { + if s.buildAborted(build) { + s.reportBuildSuperseded(slot, target.candidate) + return + } + s.log.WithError(err).WithFields(logrus.Fields{ "slot": slot, "candidate": target.candidate, @@ -852,9 +893,122 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { return } + // Publish under the build lock: a supersede that raced the build's end + // either sees the payload here (and withdraws it) or the build sees the + // abort (and never emits). + s.scheduledBuildMu.Lock() + if build.aborted { + s.scheduledBuildMu.Unlock() + s.reportBuildSuperseded(slot, target.candidate) + + return + } + + build.payload = payloadEvent + s.payloadCache.Store(payloadEvent) + s.scheduledBuildMu.Unlock() + s.emitPayloadReady(slot, payloadEvent) } +// buildAborted reports whether the build was superseded while running. +func (s *Service) buildAborted(build *candidateBuild) bool { + s.scheduledBuildMu.Lock() + defer s.scheduledBuildMu.Unlock() + + return build.aborted +} + +// reportBuildSuperseded records a superseded build as failed so its +// in-progress (or already ready) rendering is not left standing for a +// payload that will never be bid. +func (s *Service) reportBuildSuperseded(slot phase0.Slot, candidate chain.CandidateKey) { + s.log.WithFields(logrus.Fields{ + "slot": slot, + "candidate": candidate, + }).Info("Payload build superseded by beacon node attributes") + + s.buildFailedDispatcher.Fire(&PayloadBuildFailedEvent{ + Slot: slot, + Candidate: string(candidate), + Error: errBuildSuperseded, + FailedAt: time.Now(), + }) +} + +// supersedeSynthesizedBuild aborts the slot's started build on the event's +// parent tuple when that build ran from locally synthesized attributes and +// the beacon node's own attributes disagree with them. The missing-block +// fallback copies the previous slot's attributes, which are wrong whenever +// the state moved in between (an epoch transition changes the expected +// withdrawals), and the beacon node can emit its real attributes after the +// fallback fired: the in-flight engine build is cancelled, an already +// emitted payload is withdrawn from the cache so no bid commits to it, and +// the tuple is released so the caller's normal path builds again from the +// real attributes. Identical build inputs keep the running build (nothing to +// fix) and just clear its synthesized marker. Returns whether a build was +// superseded. +func (s *Service) supersedeSynthesizedBuild(event *beacon.PayloadAttributesEvent) bool { + slot := event.ProposalSlot + tuple := beacon.AttrParentKeyOf(event) + + s.scheduledBuildMu.Lock() + + state := s.slotBuilds[slot] + if state == nil { + s.scheduledBuildMu.Unlock() + return false + } + + build := state.started[tuple] + if build == nil || !build.synthesized { + s.scheduledBuildMu.Unlock() + return false + } + + if build.attrs.BuildInputsEqual(event) { + build.synthesized = false + s.scheduledBuildMu.Unlock() + + s.log.WithFields(logrus.Fields{ + "slot": slot, + "parent_hash": fmt.Sprintf("%x", event.ParentBlockHash[:8]), + }).Debug("Beacon node attributes match the synthesized build, keeping it") + + return false + } + + build.aborted = true + build.cancel() + delete(state.started, tuple) + + emitted := build.payload + if emitted != nil { + s.payloadCache.Remove(emitted) + } + s.scheduledBuildMu.Unlock() + + fields := logrus.Fields{ + "slot": slot, + "parent_hash": fmt.Sprintf("%x", event.ParentBlockHash[:8]), + "withdrawals": len(event.Withdrawals), + } + if emitted != nil { + fields["block_hash"] = fmt.Sprintf("%x", emitted.BlockHash[:8]) + } + + s.log.WithFields(fields).Warn("Beacon node attributes differ from the synthesized " + + "attributes the build ran from, aborting it and rebuilding") + + // The build goroutine reports itself while running; an already finished + // build is reported here. + if emitted != nil { + s.reportBuildSuperseded(slot, emitted.Candidate) + } + + return true +} + // applyPayloadTransform rewrites the built execution payload with the slot's // frozen jq payload transform (idempotent Freeze), in place. Because the bid // commits to the payload's block hash, Payload.BlockHash is re-synced to the @@ -980,10 +1134,8 @@ func (s *Service) handlePayloadAvailableEvent(event *beacon.PayloadAvailableEven // accounting (next_n schedule budget, slots-built stat) fires once per slot // regardless of how many candidate payloads it produced. func (s *Service) emitPayloadReady(slot phase0.Slot, payloadEvent *Payload) { - // Store in cache - s.payloadCache.Store(payloadEvent) - - // Emit the payload ready event to subscribers + // Emit the payload ready event to subscribers (the payload is already + // cached by the build that produced it). s.payloadReadyDispatcher.Fire(payloadEvent) s.log.WithFields(logrus.Fields{ diff --git a/pkg/rpc/beacon/events.go b/pkg/rpc/beacon/events.go index 525f0c4..77f2c57 100644 --- a/pkg/rpc/beacon/events.go +++ b/pkg/rpc/beacon/events.go @@ -2,6 +2,7 @@ package beacon import ( "bufio" + "bytes" "context" "encoding/hex" "encoding/json" @@ -113,6 +114,50 @@ type PayloadAttributesEvent struct { ParentBeaconBlockRoot phase0.Root TargetGasLimit uint64 InclusionListTransactions [][]byte + + // Synthesized marks an event the builder derived locally (the + // missing-block fallback copies the previous slot's attributes) instead + // of receiving it from the beacon node. Never set on node-received + // events: a node-received event for the same parent tuple supersedes a + // synthesized one. + Synthesized bool +} + +// BuildInputsEqual reports whether two events describe the same payload to +// build: every attribute the execution layer or the bid depends on matches. +// The proposal slot, the parent tuple and the informational parent block +// number (backfilled by sanitization) are not compared; callers compare +// variants of one slot and parent tuple. +func (e *PayloadAttributesEvent) BuildInputsEqual(other *PayloadAttributesEvent) bool { + if e.ProposerIndex != other.ProposerIndex || + e.Timestamp != other.Timestamp || + e.PrevRandao != other.PrevRandao || + e.SuggestedFeeRecipient != other.SuggestedFeeRecipient || + e.ParentBeaconBlockRoot != other.ParentBeaconBlockRoot || + e.TargetGasLimit != other.TargetGasLimit || + len(e.Withdrawals) != len(other.Withdrawals) || + len(e.InclusionListTransactions) != len(other.InclusionListTransactions) { + return false + } + + for i, w := range e.Withdrawals { + o := other.Withdrawals[i] + if (w == nil) != (o == nil) { + return false + } + + if w != nil && *w != *o { + return false + } + } + + for i, tx := range e.InclusionListTransactions { + if !bytes.Equal(tx, other.InclusionListTransactions[i]) { + return false + } + } + + return true } // payloadAttributesEventJSON is used for JSON unmarshaling of payload_attributes events. diff --git a/pkg/rpc/beacon/events_test.go b/pkg/rpc/beacon/events_test.go index 9b2eaf9..9a0e0dd 100644 --- a/pkg/rpc/beacon/events_test.go +++ b/pkg/rpc/beacon/events_test.go @@ -4,6 +4,8 @@ import ( "encoding/json" "testing" + "github.com/ethereum/go-ethereum/common" + "github.com/ethpandaops/go-eth2-client/spec/capella" "github.com/ethpandaops/go-eth2-client/spec/phase0" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -217,3 +219,85 @@ func TestPayloadAttributeVariants(t *testing.T) { assert.Nil(t, stream.GetLatestPayloadAttributes(20)) assert.Empty(t, stream.GetPayloadAttributesVariants(20)) } + +func TestPayloadAttributesBuildInputsEqual(t *testing.T) { + base := func() *PayloadAttributesEvent { + return &PayloadAttributesEvent{ + ProposalSlot: 30, + ProposerIndex: 7, + ParentBlockRoot: phase0.Root{0x01}, + ParentBlockNumber: 29, + ParentBlockHash: phase0.Hash32{0xaa}, + Timestamp: 1000, + PrevRandao: phase0.Root{0xcc}, + SuggestedFeeRecipient: common.Address{0x0f}, + ParentBeaconBlockRoot: phase0.Root{0x01}, + TargetGasLimit: 60_000_000, + Withdrawals: []*capella.Withdrawal{ + {Index: 1, ValidatorIndex: 10, Amount: 500}, + {Index: 2, ValidatorIndex: 11, Amount: 600}, + }, + InclusionListTransactions: [][]byte{{0x01, 0x02}}, + } + } + + tests := []struct { + name string + mutate func(e *PayloadAttributesEvent) + equal bool + }{ + {name: "identical", mutate: func(*PayloadAttributesEvent) {}, equal: true}, + { + name: "slot and parent number ignored", + mutate: func(e *PayloadAttributesEvent) { e.ProposalSlot = 31; e.ParentBlockNumber = 0; e.Synthesized = true }, + equal: true, + }, + { + name: "withdrawal amount differs", + mutate: func(e *PayloadAttributesEvent) { e.Withdrawals[1].Amount = 601 }, + equal: false, + }, + { + name: "withdrawal count differs", + mutate: func(e *PayloadAttributesEvent) { e.Withdrawals = e.Withdrawals[:1] }, + equal: false, + }, + {name: "prev randao differs", mutate: func(e *PayloadAttributesEvent) { e.PrevRandao = phase0.Root{0xdd} }, equal: false}, + {name: "timestamp differs", mutate: func(e *PayloadAttributesEvent) { e.Timestamp++ }, equal: false}, + {name: "proposer differs", mutate: func(e *PayloadAttributesEvent) { e.ProposerIndex = 8 }, equal: false}, + {name: "target gas limit differs", mutate: func(e *PayloadAttributesEvent) { e.TargetGasLimit = 0 }, equal: false}, + {name: "fee recipient differs", mutate: func(e *PayloadAttributesEvent) { e.SuggestedFeeRecipient = common.Address{} }, equal: false}, + { + name: "inclusion list differs", + mutate: func(e *PayloadAttributesEvent) { e.InclusionListTransactions = [][]byte{{0x01, 0x03}} }, + equal: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + a, b := base(), base() + tt.mutate(b) + assert.Equal(t, tt.equal, a.BuildInputsEqual(b)) + assert.Equal(t, tt.equal, b.BuildInputsEqual(a), "symmetric") + }) + } +} + +// TestInjectPayloadAttributes_NodeEventReplacesSynthesized: a node-received +// event for the same parent tuple replaces the synthesized variant, and the +// marker tells them apart. +func TestInjectPayloadAttributes_NodeEventReplacesSynthesized(t *testing.T) { + stream := NewEventStream(&Client{}) + + synthesized := &PayloadAttributesEvent{ProposalSlot: 12, ParentBlockHash: phase0.Hash32{0xaa}, Synthesized: true} + require.True(t, stream.InjectPayloadAttributes(synthesized)) + assert.True(t, stream.GetLatestPayloadAttributes(12).Synthesized) + + real := &PayloadAttributesEvent{ProposalSlot: 12, ParentBlockHash: phase0.Hash32{0xaa}, Timestamp: 5} + stream.cachePayloadAttributes(real) + + assert.Same(t, real, stream.GetLatestPayloadAttributes(12), "node event wins") + assert.Len(t, stream.GetPayloadAttributesVariants(12), 1) + assert.False(t, stream.GetLatestPayloadAttributes(12).Synthesized) +} diff --git a/pkg/slot_results/tracker.go b/pkg/slot_results/tracker.go index 7181f32..87ea7bb 100644 --- a/pkg/slot_results/tracker.go +++ b/pkg/slot_results/tracker.go @@ -591,8 +591,11 @@ func (t *Tracker) handleBuildFailed(event *payload_builder.PayloadBuildFailedEve upsertBuildOutcome(result, outcome) - // Another candidate's ready payload keeps the primary slot outcome. - if result.Build != nil && result.Build.Status == BuildStatusReady { + // Another candidate's ready payload keeps the primary slot outcome; + // the same candidate's ready payload was withdrawn (a superseded + // build), so its failure replaces it. + if result.Build != nil && result.Build.Status == BuildStatusReady && + result.Build.Candidate != event.Candidate { return } From 2b2209a0d9e5fb5f7d0451b14f62d47b8a704752 Mon Sep 17 00:00:00 2001 From: pk910 Date: Mon, 7 Sep 2026 12:21:46 +0200 Subject: [PATCH 2/4] test(e2e): run a two-node enclave so p2p bids can reach a proposer The e2e's post-Gloas p2p phase stopped winning on 2026-09-06: the ethpandaops/lodestar:glamsterdam-devnet-8 image moved to v1.47.0, which no longer inserts a beacon-API-published execution payload bid into the publishing node's own bid pool (ChainSafe/lodestar#9998). gossipsub never loops a self-published message back either, so in the single-node enclave buildoor's bids were gossiped to zero peers and node 1's proposers logged "builder_no_bid" right after "Published execution payload bid" for every slot. Run the lodestar/nethermind participant with count 2: buildoor stays wired to node 1, gossip feeds node 2's bid pool, and the p2p phase wins on a node 2 proposal (verified locally: node 2 logs "Selected builder block" for the bid node 1 published). Both validator clients carry the builder URL, so the Builder API phases keep winning on any proposer. --- .github/e2e/kurtosis.yaml | 8 ++++++++ .github/scripts/e2e-kurtosis.sh | 5 +++++ .github/workflows/_shared-e2e.yaml | 8 +++++--- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/.github/e2e/kurtosis.yaml b/.github/e2e/kurtosis.yaml index 141074a..36ba89c 100644 --- a/.github/e2e/kurtosis.yaml +++ b/.github/e2e/kurtosis.yaml @@ -1,4 +1,11 @@ participants: + # Two identical nodes. buildoor is wired to node 1 (cl-1/el-1) and publishes + # its p2p bids through node 1's beacon API, which only gossips them to peers: + # since lodestar v1.47.0 (ChainSafe/lodestar#9998) the API publish path no + # longer feeds the node's own bid pool and gossipsub never loops a + # self-published message back, so node 1's validators are blind to buildoor's + # p2p bids. The post-Gloas p2p phase is therefore won on node 2's proposals, + # whose bid pool is fed by gossip from node 1. - el_type: nethermind el_image: ethpandaops/nethermind:glamsterdam-devnet-8 cl_type: lodestar @@ -6,6 +13,7 @@ participants: vc_image: ethpandaops/lodestar:glamsterdam-devnet-8 validator_count: 32 supernode: true + count: 2 # From Gloas the external builder is configured on the VALIDATOR client and # the beacon node requests bids from it on the validator's behalf, so the # CL-side --builder.urls the package wires up only covers the pre-Gloas diff --git a/.github/scripts/e2e-kurtosis.sh b/.github/scripts/e2e-kurtosis.sh index 1701810..70d2cf4 100755 --- a/.github/scripts/e2e-kurtosis.sh +++ b/.github/scripts/e2e-kurtosis.sh @@ -234,6 +234,11 @@ jq -e '.data.PRESET_BASE == "minimal" and (.data.GLOAS_FORK_EPOCH | tonumber) == # both are active cannot tell us which one the proposer actually used. The # post-Gloas phases therefore run one flow at a time, toggled through the # settings API, and only accept wins from slots frozen after the toggle. +# +# The p2p phase needs the enclave's second node: buildoor publishes its bids +# through node 1's beacon API, which only gossips them to peers (lodestar >= +# v1.47.0 no longer feeds its own bid pool from the API path), so node 1's +# validators never see them and the win comes from a node 2 proposal. echo "== Phase 1: pre-Gloas Builder API (getHeader / blinded block)" pre_block=$(wait_for_win pre-gloas builder_api 0 $((GLOAS_SLOT - 1)) "$PREGLOAS_TIMEOUT_SECONDS") diff --git a/.github/workflows/_shared-e2e.yaml b/.github/workflows/_shared-e2e.yaml index 754e604..84354ed 100644 --- a/.github/workflows/_shared-e2e.yaml +++ b/.github/workflows/_shared-e2e.yaml @@ -11,9 +11,11 @@ jobs: e2e_kurtosis: name: Run Kurtosis E2E runs-on: ubuntu-latest - # The check runs three win phases in one enclave (pre-Gloas Builder API, - # post-Gloas p2p, post-Gloas Builder API); the post-Gloas ones first wait - # for the builder's deposit to activate on chain. + # The check runs three win phases in one two-node enclave (pre-Gloas + # Builder API, post-Gloas p2p, post-Gloas Builder API); the post-Gloas + # ones first wait for the builder's deposit to activate on chain. The + # second node is what makes the p2p phase observable at all: the bid must + # reach a proposer over gossip (see .github/e2e/kurtosis.yaml). timeout-minutes: 60 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 From 7052a393972b0a6b27767fd3831b1c4dbe2e5602 Mon Sep 17 00:00:00 2001 From: pk910 Date: Mon, 7 Sep 2026 12:36:24 +0200 Subject: [PATCH 3/4] fix(payload_builder): order superseded-build events by build sequence Review follow-ups on the supersede path: - A supersede racing the end of a build could withdraw the payload while its ready event was still being dispatched, leaving consumers on "ready" for a payload that no longer exists. The build now records when its ready dispatch finished; a supersede that finds the payload cached but not yet dispatched leaves the failure report to the build goroutine, which fires it right after the ready event. Exactly one superseded failure per build, always after that build's ready. - The aborted build's late failure could arrive after the rebuild's ready outcome for the same candidate and replace it. Every build now carries a monotonic BuildSeq on its started, ready and failed events; the slot results tracker keeps it per outcome (in memory only) and drops any event of an older build once a newer build of the same candidate has reported, while a newer build's start restarts the candidate instead of being refused as a regression. --- pkg/payload_builder/build_supersede_test.go | 71 ++++++++++++++++++- pkg/payload_builder/events.go | 5 ++ pkg/payload_builder/payload.go | 1 + pkg/payload_builder/service.go | 53 +++++++++++--- pkg/slot_results/tracker.go | 37 +++++++++- pkg/slot_results/tracker_test.go | 76 +++++++++++++++++++++ pkg/slot_results/types.go | 8 +++ 7 files changed, 239 insertions(+), 12 deletions(-) diff --git a/pkg/payload_builder/build_supersede_test.go b/pkg/payload_builder/build_supersede_test.go index de1c7d7..77446aa 100644 --- a/pkg/payload_builder/build_supersede_test.go +++ b/pkg/payload_builder/build_supersede_test.go @@ -208,7 +208,9 @@ func TestSupersedeSynthesizedBuild_WithdrawsEmittedPayload(t *testing.T) { } svc.scheduledBuildMu.Lock() + build.seq = 7 build.payload = payload + build.readyFired = true svc.payloadCache.Store(payload) svc.scheduledBuildMu.Unlock() @@ -229,11 +231,78 @@ func TestSupersedeSynthesizedBuild_WithdrawsEmittedPayload(t *testing.T) { assert.Equal(t, phase0.Slot(133), event.Slot) assert.Equal(t, string(chain.CandidateParentFull), event.Candidate) assert.Equal(t, errBuildSuperseded, event.Error) + assert.Equal(t, uint64(7), event.BuildSeq, "the failure carries the superseded build's seq") default: t.Fatal("a finished build must be reported as superseded") } } +// TestSupersedeSynthesizedBuild_FailureFollowsInFlightReady covers the window +// between caching the payload and finishing its ready dispatch: the supersede +// withdraws the payload but leaves the failure report to the build goroutine, +// so consumers always see the ready event before the superseded failure. +func TestSupersedeSynthesizedBuild_FailureFollowsInFlightReady(t *testing.T) { + spec := &chain.ChainSpec{SecondsPerSlot: 12 * time.Second, SlotsPerEpoch: 32} + svc := supersedeTestService(t, &stubChainService{spec: spec}) + + synthesized := synthesizedAttrs() + build, _ := registerBuild(svc, synthesized, true) + + payload := &Payload{ + Attributes: synthesized, + BlockHash: phase0.Hash32{0xb0, 0xd7}, + Candidate: chain.CandidateParentFull, + ReadyAt: time.Now(), + } + + // Cached, ready event still being dispatched (readyFired not yet set). + svc.scheduledBuildMu.Lock() + build.seq = 3 + build.payload = payload + svc.payloadCache.Store(payload) + svc.scheduledBuildMu.Unlock() + + failedSub := svc.SubscribePayloadBuildFailed(4, false) + defer failedSub.Unsubscribe() + + real := *synthesized + real.Synthesized = false + real.Withdrawals = []*capella.Withdrawal{{Index: 1, ValidatorIndex: 10, Amount: 104426}} + + require.True(t, svc.supersedeSynthesizedBuild(&real)) + assert.Nil(t, svc.payloadCache.Get(133), "withdrawn immediately") + + select { + case <-failedSub.Channel(): + t.Fatal("the failure must wait for the build's ready dispatch to finish") + default: + } + + // The build goroutine finishes its ready dispatch and reports the abort. + svc.completeBuild(build) + + select { + case event := <-failedSub.Channel(): + assert.Equal(t, errBuildSuperseded, event.Error) + assert.Equal(t, uint64(3), event.BuildSeq) + default: + t.Fatal("the finished ready dispatch must be followed by the superseded failure") + } + + // A build that was never aborted reports nothing on completion. + other := synthesizedAttrs() + other.ParentBlockHash = phase0.Hash32{0xbb} + otherBuild, _ := registerBuild(svc, other, false) + otherBuild.payload = &Payload{Attributes: other, Candidate: chain.CandidateParentEmpty} + svc.completeBuild(otherBuild) + + select { + case <-failedSub.Channel(): + t.Fatal("an unaborted build must not report a failure") + default: + } +} + // TestHandlePayloadAttributes_RebuildsAfterSupersede drives the whole path // through the attributes handler: the fallback's synthesized build is // aborted by the node's differing event and a fresh build starts from the @@ -265,7 +334,7 @@ func TestHandlePayloadAttributes_RebuildsAfterSupersede(t *testing.T) { // (it fails against the offline clients, which is fine: it ran). require.Eventually(t, func() bool { build := svc.startedBuild(133, key) - return build != nil && build != stale && !build.synthesized + return build != nil && build != stale && !build.synthesized && build.seq > stale.seq }, 2*time.Second, 10*time.Millisecond, "a new build from the node's attributes must start") select { diff --git a/pkg/payload_builder/events.go b/pkg/payload_builder/events.go index b341167..9a32131 100644 --- a/pkg/payload_builder/events.go +++ b/pkg/payload_builder/events.go @@ -13,6 +13,10 @@ type PayloadBuildStartedEvent struct { Slot phase0.Slot Candidate string // candidate key the build targets ("" = unclassified) StartedAt time.Time // When the build started + // BuildSeq identifies the build across its started/ready/failed events + // and orders the builds of one slot: a superseded build's late events + // carry a lower seq than its replacement's. Monotonic per process. + BuildSeq uint64 } // PayloadBuildFailedEvent is emitted when a payload build fails. Subscribers @@ -23,6 +27,7 @@ type PayloadBuildFailedEvent struct { Candidate string // candidate key the build targeted ("" = unclassified) Error string // Failure reason FailedAt time.Time // When the build failed + BuildSeq uint64 // see PayloadBuildStartedEvent.BuildSeq } // BuildSkippedEvent is emitted when the builder deliberately does not build diff --git a/pkg/payload_builder/payload.go b/pkg/payload_builder/payload.go index 6808d5d..911d356 100644 --- a/pkg/payload_builder/payload.go +++ b/pkg/payload_builder/payload.go @@ -40,6 +40,7 @@ type Payload struct { FeeRecipient common.Address // resolved proposer fee recipient for the bid BlockValue *big.Int // EL-reported block value (wei) ReadyAt time.Time // when the payload became ready + BuildSeq uint64 // see PayloadBuildStartedEvent.BuildSeq // activity is the bid/reveal log, appended by the payload_bidder and read by // the WebUI. The mutex also makes Payload copy-unsafe, enforcing the diff --git a/pkg/payload_builder/service.go b/pkg/payload_builder/service.go index b9aee41..22cb3d3 100644 --- a/pkg/payload_builder/service.go +++ b/pkg/payload_builder/service.go @@ -82,6 +82,9 @@ type Service struct { // lastBuiltSlot tracks the most recently built slot (WebUI status). lastBuiltSlot atomic.Uint64 + // buildSeq numbers every candidate build (PayloadBuildStartedEvent.BuildSeq). + buildSeq atomic.Uint64 + // EL client identification (engine_getClientVersionV1) — refreshed periodically. elClientVersionMu sync.RWMutex elClientVersion *ELClientVersion @@ -493,11 +496,13 @@ func newSlotBuildState() *slotBuildState { // beacon node's own attributes for the same tuple while it runs or after it // emitted its payload; the fields are guarded by Service.scheduledBuildMu. type candidateBuild struct { + seq uint64 // BuildSeq of the build's events attrs *beacon.PayloadAttributesEvent // attributes the build ran from synthesized bool // attrs were synthesized locally cancel context.CancelFunc // aborts the in-flight engine build aborted bool // superseded: the payload must not be emitted - payload *Payload // emitted payload (nil while building or failed) + payload *Payload // cached payload (nil while building or failed) + readyFired bool // the payload's ready event was dispatched } // errBuildSuperseded is the recorded failure reason of a build superseded by @@ -817,6 +822,7 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { defer cancel() build := &candidateBuild{ + seq: s.buildSeq.Add(1), attrs: target.attrs, synthesized: target.derived || target.attrs.Synthesized, cancel: cancel, @@ -844,12 +850,13 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { Slot: slot, Candidate: string(target.candidate), StartedAt: time.Now(), + BuildSeq: build.seq, }) payloadEvent, err := s.payloadBuilder.BuildPayloadFromAttributes(ctx, event, buildTimeMs) if err != nil { if s.buildAborted(build) { - s.reportBuildSuperseded(slot, target.candidate) + s.reportBuildSuperseded(slot, target.candidate, build.seq) return } @@ -865,11 +872,14 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { Candidate: string(target.candidate), Error: err.Error(), FailedAt: time.Now(), + BuildSeq: build.seq, }) return } + payloadEvent.BuildSeq = build.seq + // Classify the payload by the parent it was actually built on (the plan's // parent-reorg tweak may have redirected it). if event == target.attrs { @@ -888,6 +898,7 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { Candidate: string(target.candidate), Error: err.Error(), FailedAt: time.Now(), + BuildSeq: build.seq, }) return @@ -899,7 +910,7 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { s.scheduledBuildMu.Lock() if build.aborted { s.scheduledBuildMu.Unlock() - s.reportBuildSuperseded(slot, target.candidate) + s.reportBuildSuperseded(slot, target.candidate, build.seq) return } @@ -909,6 +920,23 @@ func (s *Service) executeCandidateBuild(slot phase0.Slot, target *buildTarget) { s.scheduledBuildMu.Unlock() s.emitPayloadReady(slot, payloadEvent) + s.completeBuild(build) +} + +// completeBuild marks the build's ready event as dispatched. A supersede that +// withdrew the payload while the ready event was in flight left the failure +// report to this goroutine, so the superseded failure always follows the +// same build's ready event (a consumer never ends on "ready" for a withdrawn +// payload). +func (s *Service) completeBuild(build *candidateBuild) { + s.scheduledBuildMu.Lock() + build.readyFired = true + aborted := build.aborted + s.scheduledBuildMu.Unlock() + + if aborted { + s.reportBuildSuperseded(build.payload.Attributes.ProposalSlot, build.payload.Candidate, build.seq) + } } // buildAborted reports whether the build was superseded while running. @@ -921,11 +949,14 @@ func (s *Service) buildAborted(build *candidateBuild) bool { // reportBuildSuperseded records a superseded build as failed so its // in-progress (or already ready) rendering is not left standing for a -// payload that will never be bid. -func (s *Service) reportBuildSuperseded(slot phase0.Slot, candidate chain.CandidateKey) { +// payload that will never be bid. Fired exactly once per superseded build, +// after that build's ready event when one was dispatched; the seq lets +// consumers drop it when the replacement build already reported. +func (s *Service) reportBuildSuperseded(slot phase0.Slot, candidate chain.CandidateKey, seq uint64) { s.log.WithFields(logrus.Fields{ "slot": slot, "candidate": candidate, + "build_seq": seq, }).Info("Payload build superseded by beacon node attributes") s.buildFailedDispatcher.Fire(&PayloadBuildFailedEvent{ @@ -933,6 +964,7 @@ func (s *Service) reportBuildSuperseded(slot phase0.Slot, candidate chain.Candid Candidate: string(candidate), Error: errBuildSuperseded, FailedAt: time.Now(), + BuildSeq: seq, }) } @@ -982,7 +1014,12 @@ func (s *Service) supersedeSynthesizedBuild(event *beacon.PayloadAttributesEvent build.cancel() delete(state.started, tuple) + // The build goroutine reports the abort itself while it is still running + // or still dispatching its ready event; only a fully published build is + // reported here, which keeps the failure after the ready event. emitted := build.payload + reportHere := emitted != nil && build.readyFired + if emitted != nil { s.payloadCache.Remove(emitted) } @@ -1000,10 +1037,8 @@ func (s *Service) supersedeSynthesizedBuild(event *beacon.PayloadAttributesEvent s.log.WithFields(fields).Warn("Beacon node attributes differ from the synthesized " + "attributes the build ran from, aborting it and rebuilding") - // The build goroutine reports itself while running; an already finished - // build is reported here. - if emitted != nil { - s.reportBuildSuperseded(slot, emitted.Candidate) + if reportHere { + s.reportBuildSuperseded(slot, emitted.Candidate, build.seq) } return true diff --git a/pkg/slot_results/tracker.go b/pkg/slot_results/tracker.go index 87ea7bb..087b25a 100644 --- a/pkg/slot_results/tracker.go +++ b/pkg/slot_results/tracker.go @@ -413,6 +413,7 @@ func (t *Tracker) handlePayloadReady(payload *payload_builder.Payload) { } outcome.Candidate = string(payload.Candidate) + outcome.BuildSeq = payload.BuildSeq if t.cfg.SlotArtifactCaptureEnabled && payload.ExecutionPayload != nil { idx, err := t.artifacts.StorePayload(slot, forkVersion, payload.ExecutionPayload, @@ -430,11 +431,34 @@ func (t *Tracker) handlePayloadReady(payload *payload_builder.Payload) { } t.upsert(slot, func(result *SlotResult) { + if staleBuildOutcome(result, outcome) { + return + } + upsertBuildOutcome(result, outcome) result.Build = primaryBuildOutcome(result) }) } +// staleBuildOutcome reports whether the result already holds an outcome of +// the same candidate build slot produced by a NEWER build (higher BuildSeq): +// such an outcome belongs to a superseded build whose late events must not +// overwrite its replacement's. Outcomes without a seq (legacy or baseline +// records) never count as newer. +func staleBuildOutcome(result *SlotResult, outcome *BuildOutcome) bool { + if outcome.BuildSeq == 0 { + return false + } + + for _, existing := range result.Builds { + if buildOutcomesMatch(existing, outcome) { + return existing.BuildSeq > outcome.BuildSeq + } + } + + return false +} + // buildCandidatePriority orders candidate keys from most to least canonical // for primary build selection. var buildCandidatePriority = map[string]int{ @@ -557,12 +581,16 @@ func (t *Tracker) handleBuildStarted(event *payload_builder.PayloadBuildStartedE Status: BuildStatusStarted, Candidate: event.Candidate, At: event.StartedAt, + BuildSeq: event.BuildSeq, } // Track per-candidate progress; a candidate already past started - // (ready/failed) is never regressed. + // (ready/failed) is never regressed by the same build's events. A + // NEWER build of the candidate (a rebuild after a supersede) starts + // the candidate over. for _, existing := range result.Builds { - if existing.Candidate == event.Candidate && existing.Status != BuildStatusStarted { + if existing.Candidate == event.Candidate && existing.Status != BuildStatusStarted && + !(event.BuildSeq != 0 && event.BuildSeq > existing.BuildSeq) { return } } @@ -587,6 +615,11 @@ func (t *Tracker) handleBuildFailed(event *payload_builder.PayloadBuildFailedEve Candidate: event.Candidate, Error: event.Error, At: event.FailedAt, + BuildSeq: event.BuildSeq, + } + + if staleBuildOutcome(result, outcome) { + return } upsertBuildOutcome(result, outcome) diff --git a/pkg/slot_results/tracker_test.go b/pkg/slot_results/tracker_test.go index e7e7c02..72d07f8 100644 --- a/pkg/slot_results/tracker_test.go +++ b/pkg/slot_results/tracker_test.go @@ -18,6 +18,8 @@ import ( "github.com/ethpandaops/buildoor/pkg/config" "github.com/ethpandaops/buildoor/pkg/db" "github.com/ethpandaops/buildoor/pkg/payload_bidder" + "github.com/ethpandaops/buildoor/pkg/payload_builder" + "github.com/ethpandaops/buildoor/pkg/rpc/beacon" "github.com/ethpandaops/buildoor/pkg/utils" ) @@ -562,3 +564,77 @@ func TestBuildValueFromBigInt(t *testing.T) { value := big.NewInt(1_500_000_000) require.Equal(t, "1500000000", value.String()) } + +// TestBuildOutcomeSeqOrdering: a candidate rebuilt after a supersede reports +// through two builds whose events arrive over separate subscriptions in no +// fixed order; the newer build's outcome always wins and a build's own +// superseded failure still replaces its withdrawn ready payload. +func TestBuildOutcomeSeqOrdering(t *testing.T) { + env := newTrackerTestEnv(t, false) + slot := phase0.Slot(1600) + now := time.Now() + + readyPayload := func(seq uint64, hash byte) *payload_builder.Payload { + return &payload_builder.Payload{ + Attributes: &beacon.PayloadAttributesEvent{ + ProposalSlot: slot, + ParentBlockRoot: phase0.Root{0x01}, + ParentBlockHash: phase0.Hash32{0xaa}, + }, + Candidate: chain.CandidateParentFull, + BlockHash: phase0.Hash32{hash}, + ReadyAt: now, + BuildSeq: seq, + } + } + + // Build 1 (synthesized) starts, emits a payload, then is superseded. + env.tracker.handleBuildStarted(&payload_builder.PayloadBuildStartedEvent{ + Slot: slot, Candidate: "parent_full", StartedAt: now, BuildSeq: 1, + }) + env.tracker.handlePayloadReady(readyPayload(1, 0x11)) + require.Equal(t, BuildStatusReady, env.tracker.Get(slot).Build.Status) + + env.tracker.handleBuildFailed(&payload_builder.PayloadBuildFailedEvent{ + Slot: slot, Candidate: "parent_full", Error: "superseded by beacon node attributes", + FailedAt: now, BuildSeq: 1, + }) + result := env.tracker.Get(slot) + require.Equal(t, BuildStatusFailed, result.Build.Status, "same build: failure replaces the withdrawn ready payload") + require.Len(t, result.Builds, 1) + + // Build 2 (node attributes) starts the candidate over. + env.tracker.handleBuildStarted(&payload_builder.PayloadBuildStartedEvent{ + Slot: slot, Candidate: "parent_full", StartedAt: now, BuildSeq: 2, + }) + result = env.tracker.Get(slot) + require.Len(t, result.Builds, 1) + require.Equal(t, BuildStatusStarted, result.Builds[0].Status, "a newer build restarts the candidate") + + env.tracker.handlePayloadReady(readyPayload(2, 0x22)) + result = env.tracker.Get(slot) + require.Equal(t, BuildStatusReady, result.Build.Status) + require.Equal(t, "0x2200000000000000000000000000000000000000000000000000000000000000", result.Build.BlockHash) + + // Late events of build 1 (its superseded failure, a duplicate ready) + // arrive after build 2 reported: ignored. + env.tracker.handleBuildFailed(&payload_builder.PayloadBuildFailedEvent{ + Slot: slot, Candidate: "parent_full", Error: "superseded by beacon node attributes", + FailedAt: now, BuildSeq: 1, + }) + env.tracker.handlePayloadReady(readyPayload(1, 0x11)) + env.tracker.handleBuildStarted(&payload_builder.PayloadBuildStartedEvent{ + Slot: slot, Candidate: "parent_full", StartedAt: now, BuildSeq: 1, + }) + + result = env.tracker.Get(slot) + require.Len(t, result.Builds, 1) + require.Equal(t, BuildStatusReady, result.Build.Status, "stale build 1 events never clobber build 2") + require.Equal(t, "0x2200000000000000000000000000000000000000000000000000000000000000", result.Build.BlockHash) + + // Build 2's own failure would still apply (same seq). + env.tracker.handleBuildFailed(&payload_builder.PayloadBuildFailedEvent{ + Slot: slot, Candidate: "parent_full", Error: "boom", FailedAt: now, BuildSeq: 2, + }) + require.Equal(t, BuildStatusFailed, env.tracker.Get(slot).Build.Status) +} diff --git a/pkg/slot_results/types.go b/pkg/slot_results/types.go index 2954184..d93313f 100644 --- a/pkg/slot_results/types.go +++ b/pkg/slot_results/types.go @@ -94,6 +94,14 @@ type BuildOutcome struct { Status BuildStatus `json:"status"` SkipReason string `json:"skip_reason,omitempty"` // action_plan.BuildSkipReason* when skipped + // BuildSeq is the producing build's sequence number (payload_builder + // PayloadBuildStartedEvent.BuildSeq): a slot's candidate can be built + // more than once (a superseded build and its replacement), and the + // events of the two builds arrive over separate subscriptions in no + // fixed order, so an outcome from an older build never overwrites a + // newer build's. In-memory only: seqs restart with the process. + BuildSeq uint64 `json:"-"` + // Candidate classifies which build-parent candidate this outcome belongs // to (parent_full, parent_empty, grandparent_full, grandparent_empty; // empty = unclassified or single-build slot). From f56f94fc91d4fa8766bc73b8670a91145ce15c9d Mon Sep 17 00:00:00 2001 From: pk910 Date: Mon, 7 Sep 2026 12:36:24 +0200 Subject: [PATCH 4/4] test(e2e): give the two-node enclave a 120 s genesis delay With two participants buildoor comes up ~90 s after the enclave starts, which on a GitHub runner is already slot 1. The lodestar VCs register their validators with the builder once at the start of epoch 0 and retry only at the next epoch, which is Gloas, so every pre-Gloas getHeader hit "no registration for this pubkey" and phase 1 timed out. --- .github/e2e/kurtosis.yaml | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/.github/e2e/kurtosis.yaml b/.github/e2e/kurtosis.yaml index 36ba89c..183cda4 100644 --- a/.github/e2e/kurtosis.yaml +++ b/.github/e2e/kurtosis.yaml @@ -26,7 +26,13 @@ participants: network_params: preset: minimal - genesis_delay: 20 + # buildoor is launched after both participants and must be listening before + # slot 0: the lodestar VCs register their validators with the builder once + # at the start of epoch 0 and retry only at the next epoch boundary, which + # is already Gloas (gloas_fork_epoch 1), so a late buildoor loses the whole + # pre-Gloas Builder API phase (slots 0-7). Two nodes take ~90 s to come up + # on a GitHub runner. + genesis_delay: 120 gloas_fork_epoch: 1 mev_type: buildoor