fix(abci): ignore comet ctx and use app ctx

.gitignore

@@ -231,6 +231,7 @@ coverage-test-unit-cover.txt
 coverage-merged.txt
 test-simulated.txt
 test-unit-cover.txt
+temp-test-simulated.txt
 .vercel
 
 # server dev env for Air

beacon/blockchain/payload.go

@@ -103,21 +103,10 @@ func (s *Service) forceSyncUponFinalize(
 		return err
 	}
 
-	switch err := s.executionEngine.NotifyNewPayload(ctx, payloadReq); {
-	case err == nil:
-		// Do nothing and move on to NotifyForkchoiceUpdate.
-
-	case errors.IsAny(err,
-		engineerrors.ErrSyncingPayloadStatus,
-		engineerrors.ErrAcceptedPayloadStatus):
-		// Don't return error here, because we want to send the forkchoice update regardless.
-		s.logger.Warn("pushed new payload to SYNCING node during force startup",
-			"error", err,
-			"blockNum", executionPayload.GetNumber(),
-			"blockHash", executionPayload.GetBlockHash(),
-		)
-
-	default:
+	// We set retryOnSyncingStatus to false here. We can ignore SYNCING status and proceed
+	// to the FCU.
+	err := s.executionEngine.NotifyNewPayload(ctx, payloadReq, false)
+	if err != nil {
 		return fmt.Errorf("startSyncUponFinalize NotifyNewPayload failed: %w", err)
 	}
 
@@ -132,7 +121,7 @@ func (s *Service) forceSyncUponFinalize(
 		s.chainSpec.ActiveForkVersionForSlot(beaconBlock.GetSlot()),
 	)
 
-	switch _, err := s.executionEngine.NotifyForkchoiceUpdate(ctx, req); {
+	switch _, err = s.executionEngine.NotifyForkchoiceUpdate(ctx, req); {
 	case err == nil:
 		return nil
 

beacon/blockchain/types.go

@@ -60,10 +60,11 @@ type BlobSidecars[T any] interface {
 
 // ExecutionEngine is the interface for the execution engine.
 type ExecutionEngine interface {
-	// NotifyNewPayload notifies the execution client of new payload
+	// NotifyNewPayload notifies the execution client of new payload.
 	NotifyNewPayload(
 		ctx context.Context,
 		req *ctypes.NewPayloadRequest,
+		retryOnSyncingStatus bool,
 	) error
 	// NotifyForkchoiceUpdate notifies the execution client of a forkchoice
 	// update.

cli/flags/flags.go

@@ -44,6 +44,8 @@ const (
 	engineRoot              = beaconKitRoot + "engine."
 	RPCDialURL              = engineRoot + "rpc-dial-url"
 	RPCRetries              = engineRoot + "rpc-retries"
+	RPCRetryInterval        = engineRoot + "rpc-retry-interval"
+	RPCMaxRetryInterval     = engineRoot + "rpc-max-retry-interval"
 	RPCTimeout              = engineRoot + "rpc-timeout"
 	RPCStartupCheckInterval = engineRoot + "rpc-startup-check-interval"
 	RPCHealthCheckInteval   = engineRoot + "rpc-health-check-interval"
@@ -92,6 +94,12 @@ func AddBeaconKitFlags(startCmd *cobra.Command) {
 	startCmd.Flags().Uint64(
 		RPCRetries, defaultCfg.Engine.RPCRetries, "rpc retries",
 	)
+	startCmd.Flags().Duration(
+		RPCRetryInterval, defaultCfg.Engine.RPCRetryInterval, "initial rpc retry interval",
+	)
+	startCmd.Flags().Duration(
+		RPCMaxRetryInterval, defaultCfg.Engine.RPCMaxRetryInterval, "max rpc retry interval",
+	)
 	startCmd.Flags().Duration(
 		RPCTimeout, defaultCfg.Engine.RPCTimeout, "rpc timeout",
 	)

consensus/cometbft/service/abci.go

@@ -37,19 +37,28 @@ var (
 )
 
 func (s *Service) InitChain(
-	ctx context.Context,
+	_ context.Context,
 	req *cmtabci.InitChainRequest,
 ) (*cmtabci.InitChainResponse, error) {
-	return s.initChain(ctx, req)
+	//nolint:contextcheck // see s.ctx comment for more details
+	return s.initChain(s.ctx, req)
 }
 
 // PrepareProposal implements the PrepareProposal ABCI method and returns a
 // ResponsePrepareProposal object to the client.
 func (s *Service) PrepareProposal(
-	ctx context.Context,
+	_ context.Context,
 	req *cmtabci.PrepareProposalRequest,
 ) (*cmtabci.PrepareProposalResponse, error) {
-	return s.prepareProposal(ctx, req)
+	// Check if ctx is still good. CometBFT does not check this.
+	if s.ctx.Err() != nil {
+		// If the context is getting cancelled, we are shutting down.
+		// It is ok returning an empty proposal.
+		//nolint:nilerr // explicitly allowing this case
+		return &cmtabci.PrepareProposalResponse{Txs: req.Txs}, nil
+	}
+	//nolint:contextcheck // see s.ctx comment for more details
+	return s.prepareProposal(s.ctx, req)
 }
 
 func (s *Service) Info(context.Context,
@@ -77,17 +86,33 @@ func (s *Service) Info(context.Context,
 // ProcessProposal implements the ProcessProposal ABCI method and returns a
 // ResponseProcessProposal object to the client.
 func (s *Service) ProcessProposal(
-	ctx context.Context,
+	_ context.Context,
 	req *cmtabci.ProcessProposalRequest,
 ) (*cmtabci.ProcessProposalResponse, error) {
-	return s.processProposal(ctx, req)
+	// Check if ctx is still good. CometBFT does not check this.
+	if s.ctx.Err() != nil {
+		// Node will panic on context cancel with "CONSENSUS FAILURE!!!" due to
+		// returning an error. This is expected. We do not want to accept or
+		// reject a proposal based on incomplete data.
+		// Returning PROCESS_PROPOSAL_STATUS_UNKNOWN will also result in comet panic.
+		return nil, s.ctx.Err()
+	}
+	//nolint:contextcheck // see s.ctx comment for more details
+	return s.processProposal(s.ctx, req)
 }
 
 func (s *Service) FinalizeBlock(
-	ctx context.Context,
+	_ context.Context,
 	req *cmtabci.FinalizeBlockRequest,
 ) (*cmtabci.FinalizeBlockResponse, error) {
-	return s.finalizeBlock(ctx, req)
+	// Check if ctx is still good. CometBFT does not check this.
+	if s.ctx.Err() != nil {
+		// Node will panic on context cancel with "CONSENSUS FAILURE!!!" due to error.
+		// We expect this to happen and do not want to finalize any incomplete or invalid state.
+		return nil, s.ctx.Err()
+	}
+	//nolint:contextcheck // see s.ctx comment for more details
+	return s.finalizeBlock(s.ctx, req)
 }
 
 // Commit implements the ABCI interface. It will commit all state that exists in
@@ -98,9 +123,16 @@ func (s *Service) FinalizeBlock(
 // against that height and gracefully halt if it matches the latest committed
 // height.
 func (s *Service) Commit(
-	ctx context.Context, req *cmtabci.CommitRequest,
+	_ context.Context, req *cmtabci.CommitRequest,
 ) (*cmtabci.CommitResponse, error) {
-	return s.commit(ctx, req)
+	// Check if ctx is still good. CometBFT does not check this.
+	if s.ctx.Err() != nil {
+		// Node will panic on context cancel with "CONSENSUS FAILURE!!!" due to error.
+		// We expect this to happen and do not want to commit any incomplete or invalid state.
+		return nil, s.ctx.Err()
+	}
+	//nolint:contextcheck // see s.ctx comment for more details
+	return s.commit(s.ctx, req)
 }
 
 //

consensus/cometbft/service/finalize_block.go

@@ -54,6 +54,9 @@ func (s *Service) finalizeBlockInternal(
 	// CometBFT.
 	if s.finalizeBlockState == nil {
 		s.finalizeBlockState = s.resetState(ctx)
+	} else {
+		// Preserve the CosmosSDK context while using the correct base ctx.
+		s.finalizeBlockState.SetContext(s.finalizeBlockState.Context().WithContext(ctx))
 	}
 
 	// Iterate over all raw transactions in the proposal and attempt to execute

consensus/cometbft/service/prepare_proposal.go

@@ -50,6 +50,7 @@ func (s *Service) prepareProposal(
 	// Always reset state given that PrepareProposal can timeout
 	// and be called again in a subsequent round.
 	s.prepareProposalState = s.resetState(ctx)
+	//nolint:contextcheck // ctx already passed via resetState
 	s.prepareProposalState.SetContext(
 		s.getContextForProposal(
 			s.prepareProposalState.Context(),

consensus/cometbft/service/process_proposal.go

@@ -57,6 +57,7 @@ func (s *Service) processProposal(
 		s.finalizeBlockState = s.resetState(ctx)
 	}
 
+	//nolint:contextcheck // ctx already passed via resetState
 	s.processProposalState.SetContext(
 		s.getContextForProposal(
 			s.processProposalState.Context(),

consensus/cometbft/service/service.go

@@ -96,6 +96,15 @@ type Service struct {
 	minRetainBlocks uint64
 
 	chainID string
+
+	// ctx is the context passed in for the service. CometBFT currently does
+	// not support context usage. It passes "context.TODO()" to apps that
+	// implement the ABCI++ interface, and does not provide a context that is
+	// a child context of the one the node originally provides to comet.
+	// Thus the app cannot tell when the context as been cancelled or not.
+	// TODO: We must use this as a workaround for now until CometBFT properly
+	// generates contexts that inherit from the parent context we provide.
+	ctx context.Context
 }
 
 func NewService(
@@ -165,6 +174,7 @@ func (s *Service) Start(
 		return err
 	}
 
+	s.ctx = ctx
 	s.node, err = node.NewNode(
 		ctx,
 		cfg,
@@ -220,6 +230,12 @@ func (s *Service) Stop() error {
 	return errors.Join(errs...)
 }
 
+// ResetAppCtx sets the app ctx for the service. This is used
+// primarily for the mock service.
+func (s *Service) ResetAppCtx(ctx context.Context) {
+	s.ctx = ctx
+}
+
 // Name returns the name of the cometbft.
 func (s *Service) Name() string {
 	return AppName
@@ -320,8 +336,9 @@ func (s *Service) getContextForProposal(
 		// on initialHeight. Panic appeases nilaway.
 		panic(fmt.Errorf("getContextForProposal: %w", errNilFinalizeBlockState))
 	}
-	ctx, _ = s.finalizeBlockState.Context().CacheContext()
-	return ctx
+	newCtx, _ := s.finalizeBlockState.Context().CacheContext()
+	// Preserve the CosmosSDK context while using the correct base ctx.
+	return newCtx.WithContext(ctx.Context())
 }
 
 // CreateQueryContext creates a new sdk.Context for a query, taking as args

execution/client/client.go

@@ -211,3 +211,19 @@ func (s *EngineClient) verifyChainIDAndConnection(
 	}
 	return nil
 }
+
+/* -------------------------------------------------------------------------- */
+/*                                   Getters                                  */
+/* -------------------------------------------------------------------------- */
+
+func (s *EngineClient) GetRPCRetries() uint64 {
+	return s.cfg.RPCRetries
+}
+
+func (s *EngineClient) GetRPCRetryInterval() time.Duration {
+	return s.cfg.RPCRetryInterval
+}
+
+func (s *EngineClient) GetRPCMaxRetryInterval() time.Duration {
+	return s.cfg.RPCMaxRetryInterval
+}

execution/client/config.go

@@ -28,7 +28,9 @@ import (
 
 const (
 	defaultDialURL                 = "http://localhost:8551"
-	defaultRPCRetries              = 3
+	defaultRPCRetries              = 0
+	defaultRPCRetryInterval        = 100 * time.Millisecond
+	defaultRPCMaxRetryInterval     = 10 * time.Second
 	defaultRPCTimeout              = 900 * time.Millisecond
 	defaultRPCStartupCheckInterval = 3 * time.Second
 	defaultRPCJWTRefreshInterval   = 30 * time.Second
@@ -43,6 +45,8 @@ func DefaultConfig() Config {
 	return Config{
 		RPCDialURL:              dialURL,
 		RPCRetries:              defaultRPCRetries,
+		RPCRetryInterval:        defaultRPCRetryInterval,
+		RPCMaxRetryInterval:     defaultRPCMaxRetryInterval,
 		RPCTimeout:              defaultRPCTimeout,
 		RPCStartupCheckInterval: defaultRPCStartupCheckInterval,
 		RPCJWTRefreshInterval:   defaultRPCJWTRefreshInterval,
@@ -55,9 +59,13 @@ type Config struct {
 	// RPCDialURL is the HTTP url of the execution client JSON-RPC endpoint.
 	RPCDialURL *url.ConnectionURL `mapstructure:"rpc-dial-url"`
 	// RPCRetries is the number of retries before shutting down consensus
-	// client.
+	// client. A value of 0 will retry infinitely.
 	RPCRetries uint64 `mapstructure:"rpc-retries"`
-	// RPCTimeout is the RPC timeout for execution client calls.
+	// RPCRetryInterval is the initial RPC backoff for repeated execution client calls.
+	RPCRetryInterval time.Duration `mapstructure:"rpc-retry-interval"`
+	// MaxRPCRetryInterval is the maximum RPC backoff for repeated execution client calls.
+	RPCMaxRetryInterval time.Duration `mapstructure:"rpc-max-retry-interval"`
+	// RPCTimeout is the RPC timeout for individual execution client calls.
 	RPCTimeout time.Duration `mapstructure:"rpc-timeout"`
 	// RPCStartupCheckInterval is the Interval for the startup check.
 	RPCStartupCheckInterval time.Duration `mapstructure:"rpc-startup-check-interval"`

execution/client/errors.go

@@ -38,6 +38,10 @@ var (
 	// ErrMismatchedEth1ChainID is returned when the chainID does not
 	// match the expected chain ID.
 	ErrMismatchedEth1ChainID = errors.New("mismatched chain ID")
+
+	// ErrBadConnection indicates that the http.Client was unable to
+	// establish a connection.
+	ErrBadConnection = errors.New("connection error")
 )
 
 // Handles errors received from the RPC server according to the specification.
@@ -66,11 +70,7 @@ func (s *EngineClient) handleRPCError(
 	var e jsonrpc.Error
 	ok := errors.As(err, &e)
 	if !ok || e == nil {
-		return errors.Wrapf(
-			err,
-			"got an unexpected server error in JSON-RPC response "+
-				"failed to convert from jsonrpc.Error",
-		)
+		return errors.Join(ErrBadConnection, err)
 	}
 
 	// Otherwise check for our engine errors.
@@ -119,3 +119,22 @@ func (s *EngineClient) handleRPCError(
 		return err
 	}
 }
+
+// IsFatalError defines errors that indicate a bad request or an otherwise
+// unusable client.
+func IsFatalError(err error) bool {
+	return jsonrpc.IsPreDefinedError(err) || errors.IsAny(
+		err,
+		ErrBadConnection,
+	)
+}
+
+// IsNonFatalError defines errors that should be ephemeral and can be
+// recovered from simply by retrying.
+func IsNonFatalError(err error) bool {
+	return errors.IsAny(
+		err,
+		engineerrors.ErrEngineAPITimeout,
+		http.ErrTimeout,
+	)
+}