feat(preconf): add optimistic payload build fallback, retry logic, and kurtosis devnet improvements

Author: fridrik01

beacon/blockchain/finalize_block.go

@@ -36,6 +36,29 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
+// sequencerFinalizeOptimisticBuild triggers an optimistic payload build from FinalizeBlock
+// when ProcessProposal was skipped (e.g., proposal arrived after the CometBFT propose timeout).
+func (s *Service) sequencerFinalizeOptimisticBuild(
+	ctx context.Context,
+	blk *ctypes.BeaconBlock,
+	consensusTime math.U64,
+) {
+	s.logger.Warn("ProcessProposal was skipped, triggering optimistic build from FinalizeBlock",
+		"current_slot", blk.GetSlot().Base10(),
+	)
+
+	st := s.storageBackend.StateFromContext(ctx)
+	ephemeralState := st.Protect(ctx)
+
+	buildData, err := s.preFetchBuildData(ephemeralState, consensusTime)
+	if err != nil {
+		s.logger.Error("Failed to prepare optimistic build data from FinalizeBlock", "error", err)
+		return
+	}
+
+	s.handleOptimisticPayloadBuild(ctx, buildData)
+}
+
 func (s *Service) FinalizeBlock(
 	ctx sdk.Context,
 	req *cmtabci.FinalizeBlockRequest,
@@ -80,7 +103,20 @@ func (s *Service) FinalizeBlock(
 	}
 
 	// STEP 4: Post Finalizations cleanups.
-	return valUpdates, s.PostFinalizeBlockOps(ctx, blk)
+	if err = s.PostFinalizeBlockOps(ctx, blk); err != nil {
+		return valUpdates, err
+	}
+
+	// STEP 5: Sequencer fallback — trigger optimistic build if ProcessProposal was skipped.
+	if s.preconfCfg.IsSequencer() && s.localBuilder.Enabled() {
+		skipped := !s.optimisticBuildTriggered.Load()
+		s.optimisticBuildTriggered.Store(false)
+		if skipped {
+			s.sequencerFinalizeOptimisticBuild(ctx, blk, math.U64(req.GetTime().Unix())) //#nosec: G115
+		}
+	}
+
+	return valUpdates, nil
 }
 
 func (s *Service) FinalizeSidecars(

beacon/blockchain/payload.go

@@ -133,7 +133,7 @@ func (s *Service) forceSyncUponFinalize(
 // to extract the data necessary to build the next block, whether current block is
 // being rejected or accepted. This is way there can be (and so should be)
 // a single function doing these ops. preFetchBuildData is that function.
-func (s *Service) preFetchBuildData(st *statedb.StateDB, currentTime math.U64, expectedProposer crypto.BLSPubkey) (
+func (s *Service) preFetchBuildData(st *statedb.StateDB, currentTime math.U64) (
 	*builder.RequestPayloadData,
 	error,
 ) {
@@ -201,7 +201,6 @@ func (s *Service) preFetchBuildData(st *statedb.StateDB, currentTime math.U64, e
 			FinalizedBlockHash: lph.GetParentHash(),
 		},
 		ParentProposerPubkey: parentProposerPubkey,
-		ExpectedProposer:     expectedProposer,
 	}, nil
 }
 

beacon/blockchain/process_proposal.go

@@ -303,35 +303,47 @@ func (s *Service) VerifyIncomingBlock(
 
 	// Determine if we should optimistically build a payload for the next slot.
 	var shouldBuildNextPayload bool
-	var expectedProposerPubkey crypto.BLSPubkey
 	if s.localBuilder.Enabled() {
-		if isNextBlockProposer {
+		switch {
+		case isNextBlockProposer && !s.preconfCfg.ShouldFetchFromSequencer():
+			// We're the next proposer and not fetching from sequencer,
+			// build optimistically on local EL for ourselves.
 			shouldBuildNextPayload = true
-		} else if s.preconfCfg != nil && s.preconfCfg.IsSequencer() {
-			expectedProposerPubkey, err = s.getNextProposerPubkey(state, nextProposerAddress)
+			s.logger.Info("Next proposer is this node, building optimistically",
+				"current_block_slot", blkSlot.Base10(),
+				"target_build_slot", (blkSlot + 1).Base10(),
+			)
+		case s.preconfCfg.IsSequencer():
+			// We're the sequencer, build for whitelisted validators.
+			// Only check whitelist if we can resolve the next proposer pubkey.
+			expectedProposerPubkey, err := s.getNextProposerPubkey(state, nextProposerAddress)
 			if err != nil {
 				s.logger.Error("Failed to get next proposer pubkey", "error", err)
 			} else {
 				shouldBuildNextPayload = s.preconfWhitelist.IsWhitelisted(expectedProposerPubkey)
-				if shouldBuildNextPayload {
-					s.logger.Info("Sequencer mode: next proposer is whitelisted, triggering optimistic build")
-				}
 			}
+			s.logger.Info("Sequencer mode: determined next proposer",
+				"current_block_slot", blkSlot.Base10(),
+				"target_build_slot", (blkSlot + 1).Base10(),
+				"next_proposer_address", fmt.Sprintf("%x", nextProposerAddress),
+				"expected_proposer_pubkey", expectedProposerPubkey.String(),
+				"is_whitelisted", shouldBuildNextPayload,
+			)
 		}
 	}
 
 	var nextBlockData *builder.RequestPayloadData
 	if shouldBuildNextPayload {
 		// makes sure that preFetchBuildData does not affect state
 		ephemeralState := state.Protect(ctx)
-		nextBlockData, err = s.preFetchBuildData(ephemeralState, blk.GetConsensusTime(), expectedProposerPubkey)
+		nextBlockData, err = s.preFetchBuildData(ephemeralState, blk.GetConsensusTime())
 		if err != nil {
 			// We don't return with err if pre-fetch fails. Instead we log the issue
 			// and still move to process the current block. Next block can always be
 			// built right after current height is finalized.
 			s.logger.Warn(
 				"Failed pre fetching data for optimistic block building",
-				"case", "block rejectiong",
+				"case", "block rejection",
 				"err", err,
 			)
 		}
@@ -365,7 +377,7 @@ func (s *Service) VerifyIncomingBlock(
 	if shouldBuildNextPayload {
 		// makes sure that preFetchBuildDataForSuccess does not affect state
 		ephemeralState := state.Protect(ctx)
-		nextBlockData, err = s.preFetchBuildData(ephemeralState, blk.GetConsensusTime(), expectedProposerPubkey)
+		nextBlockData, err = s.preFetchBuildData(ephemeralState, blk.GetConsensusTime())
 		if err != nil {
 			// We don't mark the block as rejected if it is valid but pre-fetch fails.
 			// Instead we log the issue and move to process the current block.
@@ -377,6 +389,7 @@ func (s *Service) VerifyIncomingBlock(
 			)
 			return valUpdates, nil
 		}
+		s.optimisticBuildTriggered.Store(true)
 		go s.handleOptimisticPayloadBuild(ctx, nextBlockData)
 	}
 

beacon/blockchain/service.go

@@ -80,6 +80,12 @@ type Service struct {
 	// - Validator: checks if self is whitelisted to fetch payload from sequencer
 	// Can be nil if preconf is disabled.
 	preconfWhitelist preconf.Whitelist
+
+	// optimisticBuildTriggered tracks whether ProcessProposal triggered an
+	// optimistic payload build for the next slot. FinalizeBlock checks this
+	// flag: if ProcessProposal was skipped (e.g., late proposal), the flag
+	// remains false and FinalizeBlock triggers the build as a fallback.
+	optimisticBuildTriggered atomic.Bool
 }
 
 // NewService creates a new validator service.

beacon/preconf/client.go

@@ -33,6 +33,7 @@ import (
 	ctypes "github.com/berachain/beacon-kit/consensus-types/types"
 	"github.com/berachain/beacon-kit/errors"
 	"github.com/berachain/beacon-kit/log"
+	"github.com/berachain/beacon-kit/primitives/common"
 	"github.com/berachain/beacon-kit/primitives/math"
 	"github.com/berachain/beacon-kit/primitives/net/jwt"
 )
@@ -82,13 +83,14 @@ func NewClient(
 	}
 }
 
-// GetPayloadBySlot fetches a payload from the sequencer for the given slot.
+// GetPayloadBySlot fetches a payload from the sequencer for the given slot and parent block root.
 func (c *Client) GetPayloadBySlot(
 	ctx context.Context,
 	slot math.Slot,
+	parentBlockRoot common.Root,
 ) (ctypes.BuiltExecutionPayloadEnv, error) {
 	// Build request
-	reqBody := GetPayloadRequest{Slot: slot}
+	reqBody := GetPayloadRequest{Slot: slot, ParentBlockRoot: parentBlockRoot}
 	reqJSON, err := json.Marshal(reqBody)
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal request: %w", err)
@@ -116,8 +118,7 @@ func (c *Client) GetPayloadBySlot(
 
 	resp, err := c.httpClient.Do(req)
 	if err != nil {
-		c.logger.Warn("Failed to contact sequencer", "error", err)
-		return nil, errors.Wrapf(ErrSequencerUnavailable, "request failed: %w", err)
+		return nil, errors.Wrapf(ErrSequencerUnavailable, "request failed: %v", err)
 	}
 	if resp == nil || resp.Body == nil {
 		return nil, errors.New("received nil response from sequencer")
@@ -148,9 +149,16 @@ func (c *Client) GetPayloadBySlot(
 		return nil, fmt.Errorf("failed to parse response: %w", err)
 	}
 
+	var blockHash interface{}
+	if payloadResp.ExecutionPayload != nil {
+		blockHash = payloadResp.ExecutionPayload.GetBlockHash()
+	} else {
+		blockHash = "nil"
+	}
+
 	c.logger.Info("Successfully fetched payload from sequencer",
 		"slot", slot,
-		"block_hash", payloadResp.ExecutionPayload.GetBlockHash(),
+		"block_hash", blockHash,
 	)
 
 	return payloadResp.ToExecutionPayloadEnvelope(), nil

beacon/preconf/config.go

@@ -26,7 +26,7 @@ const (
 	// DefaultAPIPort is the default port for the preconf API server.
 	DefaultAPIPort = 9090
 
-	// DefaultFetchTimeout is the default timeout for fetching payloads from sequencer.
+	// DefaultFetchTimeout is the default timeout for the HTTP client when fetching payloads from sequencer.
 	DefaultFetchTimeout = 500 * time.Millisecond
 )
 
@@ -80,10 +80,10 @@ func DefaultConfig() Config {
 
 // IsSequencer returns true if this node is configured as the sequencer.
 func (c *Config) IsSequencer() bool {
-	return c.Enabled && c.SequencerMode
+	return c != nil && c.Enabled && c.SequencerMode
 }
 
 // ShouldFetchFromSequencer returns true if this node should fetch payloads from sequencer.
 func (c *Config) ShouldFetchFromSequencer() bool {
-	return c.Enabled && c.SequencerURL != ""
+	return c != nil && c.Enabled && c.SequencerURL != ""
 }

beacon/preconf/server.go

@@ -32,6 +32,7 @@ import (
 	ctypes "github.com/berachain/beacon-kit/consensus-types/types"
 	"github.com/berachain/beacon-kit/errors"
 	"github.com/berachain/beacon-kit/log"
+	"github.com/berachain/beacon-kit/primitives/common"
 	"github.com/berachain/beacon-kit/primitives/crypto"
 	"github.com/berachain/beacon-kit/primitives/math"
 	"github.com/berachain/beacon-kit/primitives/net/jwt"
@@ -40,7 +41,7 @@ import (
 
 const (
 	// jwtValidityWindow is the time window for JWT validity (iat claim).
-	jwtValidityWindow = 60 * time.Second
+	jwtValidityWindow = 5 * time.Minute
 
 	// serverShutdownTimeout is the timeout for graceful server shutdown.
 	serverShutdownTimeout = 5 * time.Second
@@ -52,12 +53,10 @@ const (
 	authHeaderParts = 2
 )
 
-// PayloadProvider is an interface for retrieving payloads by slot.
+// PayloadProvider is an interface for retrieving payloads by slot and parent block root.
 type PayloadProvider interface {
-	// GetPayloadBySlot returns the payload for the given slot if available.
-	GetPayloadBySlot(ctx context.Context, slot math.Slot) (ctypes.BuiltExecutionPayloadEnv, error)
-	// GetExpectedProposer returns the expected proposer for the given slot.
-	GetExpectedProposer(slot math.Slot) (crypto.BLSPubkey, bool)
+	// GetPayloadBySlot returns the payload for the given slot and parent block root if available.
+	GetPayloadBySlot(ctx context.Context, slot math.Slot, parentBlockRoot common.Root) (ctypes.BuiltExecutionPayloadEnv, error)
 }
 
 // Server is the preconf API server that serves GetPayload requests from validators.
@@ -110,7 +109,12 @@ func (s *Server) Start(_ context.Context) error {
 	s.httpServer = server
 	s.mu.Unlock()
 
-	s.logger.Info("Starting preconf API server", "address", addr)
+	s.logger.Info("Starting preconf API server", "address", addr, "num_validator_jwts", len(s.validatorJWTs))
+
+	// Log the registered validator pubkeys for debugging
+	for pubkey := range s.validatorJWTs {
+		s.logger.Info("Registered validator JWT", "pubkey", pubkey.String())
+	}
 
 	go func() {
 		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
@@ -169,45 +173,33 @@ func (s *Server) handleGetPayload(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Validate that the requesting validator is the expected proposer for this slot
-	expectedProposer, found := s.payloadProvider.GetExpectedProposer(req.Slot)
-	if !found {
-		s.logger.Warn("No expected proposer found for slot", "slot", req.Slot)
-		s.writeError(w, http.StatusNotFound, "no payload building in progress for slot")
-		return
-	}
-	if expectedProposer != pubkey {
-		s.logger.Warn("Validator is not the expected proposer",
-			"slot", req.Slot,
-			"expected", expectedProposer.String()[:16]+"...",
-			"actual", pubkey.String()[:16]+"...",
-		)
-		s.writeError(w, http.StatusForbidden, "validator is not the expected proposer for this slot")
-		return
-	}
+	s.logger.Info("Preconf server received payload request",
+		"slot", req.Slot,
+		"validator_pubkey", pubkey.String(),
+	)
 
 	// Get the payload from provider
 	ctx := r.Context()
-	envelope, err := s.payloadProvider.GetPayloadBySlot(ctx, req.Slot)
+	startTime := time.Now()
+	envelope, err := s.payloadProvider.GetPayloadBySlot(ctx, req.Slot, req.ParentBlockRoot)
+	elapsed := time.Since(startTime)
 	if err != nil {
-		s.logger.Warn("Failed to get payload", "slot", req.Slot, "error", err)
+		s.logger.Warn("Failed to get payload",
+			"slot", req.Slot,
+			"error", err,
+			"elapsed", elapsed,
+		)
 		s.writeError(w, http.StatusNotFound, "payload not available: "+err.Error())
 		return
 	}
 
-	// Convert to response
-	resp := NewGetPayloadResponseFromEnvelope(envelope)
-
-	s.logger.Info("Serving payload to validator",
-		"slot", req.Slot,
-		"validator", pubkey.String()[:16]+"...",
-	)
-
 	// Write response
 	w.Header().Set("Content-Type", "application/json")
-	if err = json.NewEncoder(w).Encode(resp); err != nil {
+	if err = json.NewEncoder(w).Encode(NewGetPayloadResponseFromEnvelope(envelope)); err != nil {
 		s.logger.Error("Failed to encode response", "error", err)
 	}
+
+	s.logger.Info("GetPayloadBySlot completed", "slot", req.Slot, "elapsed", elapsed)
 }
 
 // validateJWT validates the JWT token from the Authorization header and returns