Defer application of http.TimeoutHandler until endpoint middleware so specific endpoints can opt out

Author: jshufro

api/server/httprest/server.go

@@ -48,7 +48,6 @@ func New(ctx context.Context, opts ...Option) (*Server, error) {
 	handler = middleware.MiddlewareChain(g.cfg.router, g.cfg.middlewares)
 	if g.cfg.timeout > 0*time.Second {
 		defaultReadHeaderTimeout = g.cfg.timeout
-		handler = http.TimeoutHandler(handler, g.cfg.timeout, "request timed out")
 	}
 	g.server = &http.Server{
 		Addr:              g.cfg.httpAddr,

beacon-chain/node/node.go

@@ -938,6 +938,7 @@ func (b *BeaconNode) registerRPCService(router *http.ServeMux) error {
 	mockEth1DataVotes := b.cliCtx.Bool(flags.InteropMockEth1DataVotesFlag.Name)
 	maxMsgSize := b.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
 	enableDebugRPCEndpoints := !b.cliCtx.Bool(flags.DisableDebugRPCEndpoints.Name)
+	apiTimeout := b.cliCtx.Duration(cmd.ApiTimeoutFlag.Name)
 
 	p2pService := b.fetchP2P()
 	rpcService := rpc.NewService(b.ctx, &rpc.Config{
@@ -994,6 +995,7 @@ func (b *BeaconNode) registerRPCService(router *http.ServeMux) error {
 		TrackedValidatorsCache:    b.trackedValidatorsCache,
 		PayloadIDCache:            b.payloadIDCache,
 		LCStore:                   b.lcStore,
+		ApiTimeout:                apiTimeout,
 	})
 
 	return b.services.RegisterService(rpcService)

beacon-chain/rpc/endpoints.go

@@ -33,6 +33,9 @@ type endpoint struct {
 	middleware []middleware.Middleware
 	handler    http.HandlerFunc
 	methods    []string
+
+	errorMiddlewareIncompatible bool
+	timeoutHandlerIncompatible  bool
 }
 
 // responseWriter is the wrapper to http Response writer.
@@ -63,9 +66,9 @@ func (e *endpoint) handlerWithMiddleware() http.HandlerFunc {
 	)
 
 	return func(w http.ResponseWriter, r *http.Request) {
-		// SSE errors are handled separately to avoid interference with the streaming
-		// mechanism and ensure accurate error tracking.
-		if e.template == "/eth/v1/events" {
+		// Some handlers (i.e. SSE) handle errors separately to avoid interference with
+		// the streaming mechanism and ensure accurate error tracking.
+		if e.errorMiddlewareIncompatible {
 			handler.ServeHTTP(w, r)
 			return
 		}
@@ -1164,6 +1167,12 @@ func (s *Service) eventsEndpoints() []endpoint {
 			},
 			handler: server.StreamEvents,
 			methods: []string{http.MethodGet},
+
+			// events endpoint handles errors separately
+			errorMiddlewareIncompatible: true,
+			// http.TimeoutHandler causes this handler to behave incorrectly, since it uses SetDeadline
+			// on the underlying net/http response
+			timeoutHandlerIncompatible: true,
 		},
 	}
 }

beacon-chain/rpc/service.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/blockchain"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/builder"
@@ -125,6 +126,7 @@ type Config struct {
 	TrackedValidatorsCache    *cache.TrackedValidatorsCache
 	PayloadIDCache            *cache.PayloadIDCache
 	LCStore                   *lightClient.Store
+	ApiTimeout                time.Duration
 }
 
 // NewService instantiates a new RPC service instance that will
@@ -301,9 +303,13 @@ func NewService(ctx context.Context, cfg *Config) *Service {
 	endpoints := s.endpoints(s.cfg.EnableDebugRPCEndpoints, blocker, stater, rewardFetcher, validatorServer, coreService, ch)
 	for _, e := range endpoints {
 		for i := range e.methods {
+			handler := e.handlerWithMiddleware()
+			if s.cfg.ApiTimeout > 0 && !e.timeoutHandlerIncompatible {
+				handler = http.TimeoutHandler(handler, s.cfg.ApiTimeout, "request timed out").ServeHTTP
+			}
 			s.cfg.Router.HandleFunc(
 				fmt.Sprintf("%s %s", e.methods[i], e.template),
-				e.handlerWithMiddleware(),
+				handler,
 			)
 		}
 	}

changelog/jshufro_fix_api_timeout_event_handler.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Fixed /eth/v1/events incompatibility with --api-timeout argument