Add profile to sync block path

* Increase health tracker channel buffer size

* keep only the latest value

* Make health test blocking as a regression test for PR #14807

* Fix new race conditions in the MockHealthClient

---------

Co-authored-by: Preston Van Loon <preston@pvl.dev>

api/client/beacon/health.go

@@ -48,8 +48,13 @@ func (n *NodeHealthTracker) CheckHealth(ctx context.Context) bool {
 	if isStatusChanged {
 		// Update the health status
 		n.isHealthy = &newStatus
-		// Send the new status to the health channel
-		n.healthChan <- newStatus
+		// Send the new status to the health channel, potentially overwriting the existing value
+		select {
+		case <-n.healthChan:
+			n.healthChan <- newStatus
+		default:
+			n.healthChan <- newStatus
+		}
 	}
 	return newStatus
 }

api/client/beacon/health_test.go

@@ -87,12 +87,6 @@ func TestNodeHealth_Concurrency(t *testing.T) {
 	// Number of goroutines to spawn for both reading and writing
 	numGoroutines := 6
 
-	go func() {
-		for range n.HealthUpdates() {
-			// Consume values to avoid blocking on channel send.
-		}
-	}()
-
 	wg.Add(numGoroutines * 2) // for readers and writers
 
 	// Concurrently update health status

api/client/beacon/testing/mock.go

@@ -3,6 +3,7 @@ package testing
 import (
 	"context"
 	"reflect"
+	"sync"
 
 	"github.com/prysmaticlabs/prysm/v5/api/client/beacon/iface"
 	"go.uber.org/mock/gomock"
@@ -16,6 +17,7 @@ var (
 type MockHealthClient struct {
 	ctrl     *gomock.Controller
 	recorder *MockHealthClientMockRecorder
+	sync.Mutex
 }
 
 // MockHealthClientMockRecorder is the mock recorder for MockHealthClient.
@@ -25,6 +27,8 @@ type MockHealthClientMockRecorder struct {
 
 // IsHealthy mocks base method.
 func (m *MockHealthClient) IsHealthy(arg0 context.Context) bool {
+	m.Lock()
+	defer m.Unlock()
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "IsHealthy", arg0)
 	ret0, ok := ret[0].(bool)
@@ -41,6 +45,8 @@ func (m *MockHealthClient) EXPECT() *MockHealthClientMockRecorder {
 
 // IsHealthy indicates an expected call of IsHealthy.
 func (mr *MockHealthClientMockRecorder) IsHealthy(arg0 any) *gomock.Call {
+	mr.mock.Lock()
+	defer mr.mock.Unlock()
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsHealthy", reflect.TypeOf((*MockHealthClient)(nil).IsHealthy), arg0)
 }

api/gateway/BUILD.bazel

@@ -14,7 +14,7 @@ go_library(
         "//validator:__subpackages__",
     ],
     deps = [
-        "//api/server:go_default_library",
+        "//api/server/middleware:go_default_library",
         "//runtime:go_default_library",
         "@com_github_gorilla_mux//:go_default_library",
         "@com_github_grpc_ecosystem_grpc_gateway_v2//runtime:go_default_library",

api/gateway/gateway.go

@@ -11,7 +11,7 @@ import (
 	"github.com/gorilla/mux"
 	gwruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/pkg/errors"
-	"github.com/prysmaticlabs/prysm/v5/api/server"
+	"github.com/prysmaticlabs/prysm/v5/api/server/middleware"
 	"github.com/prysmaticlabs/prysm/v5/runtime"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
@@ -104,7 +104,7 @@ func (g *Gateway) Start() {
 		}
 	}
 
-	corsMux := server.CorsHandler(g.cfg.allowedOrigins).Middleware(g.cfg.router)
+	corsMux := middleware.CorsHandler(g.cfg.allowedOrigins).Middleware(g.cfg.router)
 
 	if g.cfg.muxHandler != nil {
 		g.cfg.router.PathPrefix("/").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

api/server/BUILD.bazel

@@ -2,29 +2,14 @@ load("@prysm//tools/go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = [
-        "error.go",
-        "middleware.go",
-        "util.go",
-    ],
+    srcs = ["error.go"],
     importpath = "github.com/prysmaticlabs/prysm/v5/api/server",
     visibility = ["//visibility:public"],
-    deps = [
-        "@com_github_gorilla_mux//:go_default_library",
-        "@com_github_rs_cors//:go_default_library",
-    ],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = [
-        "error_test.go",
-        "middleware_test.go",
-        "util_test.go",
-    ],
+    srcs = ["error_test.go"],
     embed = [":go_default_library"],
-    deps = [
-        "//testing/assert:go_default_library",
-        "//testing/require:go_default_library",
-    ],
+    deps = ["//testing/assert:go_default_library"],
 )

api/server/middleware.go

@@ -1,32 +0,0 @@
-package server
-
-import (
-	"net/http"
-
-	"github.com/gorilla/mux"
-	"github.com/rs/cors"
-)
-
-// NormalizeQueryValuesHandler normalizes an input query of "key=value1,value2,value3" to "key=value1&key=value2&key=value3"
-func NormalizeQueryValuesHandler(next http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		query := r.URL.Query()
-		NormalizeQueryValues(query)
-		r.URL.RawQuery = query.Encode()
-
-		next.ServeHTTP(w, r)
-	})
-}
-
-// CorsHandler sets the cors settings on api endpoints
-func CorsHandler(allowOrigins []string) mux.MiddlewareFunc {
-	c := cors.New(cors.Options{
-		AllowedOrigins:   allowOrigins,
-		AllowedMethods:   []string{http.MethodPost, http.MethodGet, http.MethodDelete, http.MethodOptions},
-		AllowCredentials: true,
-		MaxAge:           600,
-		AllowedHeaders:   []string{"*"},
-	})
-
-	return c.Handler
-}

api/server/middleware/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@prysm//tools/go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "middleware.go",
+        "util.go",
+    ],
+    importpath = "github.com/prysmaticlabs/prysm/v5/api/server/middleware",
+    visibility = ["//visibility:public"],
+    deps = [
+        "@com_github_gorilla_mux//:go_default_library",
+        "@com_github_rs_cors//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "middleware_test.go",
+        "util_test.go",
+    ],
+    embed = [":go_default_library"],
+    deps = [
+        "//api:go_default_library",
+        "//testing/assert:go_default_library",
+        "//testing/require:go_default_library",
+    ],
+)

api/server/middleware/middleware.go

@@ -0,0 +1,112 @@
+package middleware
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/gorilla/mux"
+	"github.com/rs/cors"
+)
+
+// NormalizeQueryValuesHandler normalizes an input query of "key=value1,value2,value3" to "key=value1&key=value2&key=value3"
+func NormalizeQueryValuesHandler(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		query := r.URL.Query()
+		NormalizeQueryValues(query)
+		r.URL.RawQuery = query.Encode()
+
+		next.ServeHTTP(w, r)
+	})
+}
+
+// CorsHandler sets the cors settings on api endpoints
+func CorsHandler(allowOrigins []string) mux.MiddlewareFunc {
+	c := cors.New(cors.Options{
+		AllowedOrigins:   allowOrigins,
+		AllowedMethods:   []string{http.MethodPost, http.MethodGet, http.MethodDelete, http.MethodOptions},
+		AllowCredentials: true,
+		MaxAge:           600,
+		AllowedHeaders:   []string{"*"},
+	})
+
+	return c.Handler
+}
+
+// ContentTypeHandler checks request for the appropriate media types otherwise returning a http.StatusUnsupportedMediaType error
+func ContentTypeHandler(acceptedMediaTypes []string) mux.MiddlewareFunc {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			// skip the GET request
+			if r.Method == http.MethodGet {
+				next.ServeHTTP(w, r)
+				return
+			}
+			contentType := r.Header.Get("Content-Type")
+			if contentType == "" {
+				http.Error(w, "Content-Type header is missing", http.StatusUnsupportedMediaType)
+				return
+			}
+
+			accepted := false
+			for _, acceptedType := range acceptedMediaTypes {
+				if strings.TrimSpace(contentType) == strings.TrimSpace(acceptedType) {
+					accepted = true
+					break
+				}
+			}
+
+			if !accepted {
+				http.Error(w, fmt.Sprintf("Unsupported media type: %s", contentType), http.StatusUnsupportedMediaType)
+				return
+			}
+
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
+// AcceptHeaderHandler checks if the client's response preference is handled
+func AcceptHeaderHandler(serverAcceptedTypes []string) mux.MiddlewareFunc {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			acceptHeader := r.Header.Get("Accept")
+			// header is optional and should skip if not provided
+			if acceptHeader == "" {
+				next.ServeHTTP(w, r)
+				return
+			}
+
+			accepted := false
+			acceptTypes := strings.Split(acceptHeader, ",")
+			// follows rules defined in https://datatracker.ietf.org/doc/html/rfc2616#section-14.1
+			for _, acceptType := range acceptTypes {
+				acceptType = strings.TrimSpace(acceptType)
+				if acceptType == "*/*" {
+					accepted = true
+					break
+				}
+				for _, serverAcceptedType := range serverAcceptedTypes {
+					if strings.HasPrefix(acceptType, serverAcceptedType) {
+						accepted = true
+						break
+					}
+					if acceptType != "/*" && strings.HasSuffix(acceptType, "/*") && strings.HasPrefix(serverAcceptedType, acceptType[:len(acceptType)-2]) {
+						accepted = true
+						break
+					}
+				}
+				if accepted {
+					break
+				}
+			}
+
+			if !accepted {
+				http.Error(w, fmt.Sprintf("Not Acceptable: %s", acceptHeader), http.StatusNotAcceptable)
+				return
+			}
+
+			next.ServeHTTP(w, r)
+		})
+	}
+}

api/server/middleware/middleware_test.go

@@ -0,0 +1,199 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/prysmaticlabs/prysm/v5/api"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+)
+
+func TestNormalizeQueryValuesHandler(t *testing.T) {
+	nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte("next handler"))
+		require.NoError(t, err)
+	})
+
+	handler := NormalizeQueryValuesHandler(nextHandler)
+
+	tests := []struct {
+		name          string
+		inputQuery    string
+		expectedQuery string
+	}{
+		{
+			name:          "3 values",
+			inputQuery:    "key=value1,value2,value3",
+			expectedQuery: "key=value1&key=value2&key=value3", // replace with expected normalized value
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			req, err := http.NewRequest("GET", "/test?"+test.inputQuery, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			rr := httptest.NewRecorder()
+			handler.ServeHTTP(rr, req)
+
+			if rr.Code != http.StatusOK {
+				t.Errorf("handler returned wrong status code: got %v want %v", rr.Code, http.StatusOK)
+			}
+
+			if req.URL.RawQuery != test.expectedQuery {
+				t.Errorf("query not normalized: got %v want %v", req.URL.RawQuery, test.expectedQuery)
+			}
+
+			if rr.Body.String() != "next handler" {
+				t.Errorf("next handler was not executed")
+			}
+		})
+	}
+}
+
+func TestContentTypeHandler(t *testing.T) {
+	acceptedMediaTypes := []string{api.JsonMediaType, api.OctetStreamMediaType}
+
+	nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte("next handler"))
+		require.NoError(t, err)
+	})
+
+	handler := ContentTypeHandler(acceptedMediaTypes)(nextHandler)
+
+	tests := []struct {
+		name               string
+		contentType        string
+		expectedStatusCode int
+		isGet              bool
+	}{
+		{
+			name:               "Accepted Content-Type - application/json",
+			contentType:        api.JsonMediaType,
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Accepted Content-Type - ssz format",
+			contentType:        api.OctetStreamMediaType,
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Unsupported Content-Type - text/plain",
+			contentType:        "text/plain",
+			expectedStatusCode: http.StatusUnsupportedMediaType,
+		},
+		{
+			name:               "Missing Content-Type",
+			contentType:        "",
+			expectedStatusCode: http.StatusUnsupportedMediaType,
+		},
+		{
+			name:               "GET request skips content type check",
+			contentType:        "",
+			expectedStatusCode: http.StatusOK,
+			isGet:              true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			httpMethod := http.MethodPost
+			if tt.isGet {
+				httpMethod = http.MethodGet
+			}
+			req := httptest.NewRequest(httpMethod, "/", nil)
+			if tt.contentType != "" {
+				req.Header.Set("Content-Type", tt.contentType)
+			}
+			rr := httptest.NewRecorder()
+
+			handler.ServeHTTP(rr, req)
+
+			if status := rr.Code; status != tt.expectedStatusCode {
+				t.Errorf("handler returned wrong status code: got %v want %v", status, tt.expectedStatusCode)
+			}
+		})
+	}
+}
+
+func TestAcceptHeaderHandler(t *testing.T) {
+	acceptedTypes := []string{"application/json", "application/octet-stream"}
+
+	nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte("next handler"))
+		require.NoError(t, err)
+	})
+
+	handler := AcceptHeaderHandler(acceptedTypes)(nextHandler)
+
+	tests := []struct {
+		name               string
+		acceptHeader       string
+		expectedStatusCode int
+	}{
+		{
+			name:               "Accepted Accept-Type - application/json",
+			acceptHeader:       "application/json",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Accepted Accept-Type - application/octet-stream",
+			acceptHeader:       "application/octet-stream",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Accepted Accept-Type with parameters",
+			acceptHeader:       "application/json;q=0.9, application/octet-stream;q=0.8",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "Unsupported Accept-Type - text/plain",
+			acceptHeader:       "text/plain",
+			expectedStatusCode: http.StatusNotAcceptable,
+		},
+		{
+			name:               "Missing Accept header",
+			acceptHeader:       "",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "*/* is accepted",
+			acceptHeader:       "*/*",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "application/* is accepted",
+			acceptHeader:       "application/*",
+			expectedStatusCode: http.StatusOK,
+		},
+		{
+			name:               "/* is unsupported",
+			acceptHeader:       "/*",
+			expectedStatusCode: http.StatusNotAcceptable,
+		},
+		{
+			name:               "application/ is unsupported",
+			acceptHeader:       "application/",
+			expectedStatusCode: http.StatusNotAcceptable,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest("GET", "/", nil)
+			if tt.acceptHeader != "" {
+				req.Header.Set("Accept", tt.acceptHeader)
+			}
+			rr := httptest.NewRecorder()
+
+			handler.ServeHTTP(rr, req)
+
+			if status := rr.Code; status != tt.expectedStatusCode {
+				t.Errorf("handler returned wrong status code: got %v want %v", status, tt.expectedStatusCode)
+			}
+		})
+	}
+}

api/server/middleware/util.go

@@ -1,4 +1,4 @@
-package server
+package middleware
 
 import (
 	"net/url"

api/server/middleware/util_test.go

@@ -1,4 +1,4 @@
-package server
+package middleware
 
 import (
 	"testing"

api/server/middleware_test.go

@@ -1,54 +0,0 @@
-package server
-
-import (
-	"net/http"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/prysmaticlabs/prysm/v5/testing/require"
-)
-
-func TestNormalizeQueryValuesHandler(t *testing.T) {
-	nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		_, err := w.Write([]byte("next handler"))
-		require.NoError(t, err)
-	})
-
-	handler := NormalizeQueryValuesHandler(nextHandler)
-
-	tests := []struct {
-		name          string
-		inputQuery    string
-		expectedQuery string
-	}{
-		{
-			name:          "3 values",
-			inputQuery:    "key=value1,value2,value3",
-			expectedQuery: "key=value1&key=value2&key=value3", // replace with expected normalized value
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			req, err := http.NewRequest("GET", "/test?"+test.inputQuery, nil)
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			rr := httptest.NewRecorder()
-			handler.ServeHTTP(rr, req)
-
-			if rr.Code != http.StatusOK {
-				t.Errorf("handler returned wrong status code: got %v want %v", rr.Code, http.StatusOK)
-			}
-
-			if req.URL.RawQuery != test.expectedQuery {
-				t.Errorf("query not normalized: got %v want %v", req.URL.RawQuery, test.expectedQuery)
-			}
-
-			if rr.Body.String() != "next handler" {
-				t.Errorf("next handler was not executed")
-			}
-		})
-	}
-}

beacon-chain/core/blocks/attestation_benchmark_test.go

@@ -1,254 +0,0 @@
-package blocks
-
-import (
-	"bytes"
-	"encoding/binary"
-	"sync"
-	"testing"
-
-	"github.com/prysmaticlabs/go-bitfield"
-	lruwrpr "github.com/prysmaticlabs/prysm/v5/cache/lru"
-	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
-	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
-	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
-	"github.com/prysmaticlabs/prysm/v5/testing/assert"
-)
-
-func makeAtts(b *testing.B) []*ethpb.Attestation {
-	attData := make([]*ethpb.AttestationData, 0, 400)
-	for i := 0; i < 400; i++ {
-		bRoot := bytesutil.Bytes32(20)
-		sRoot := bytesutil.Bytes32(1000)
-		tRoot := bytesutil.Bytes32(2000)
-		ad := &ethpb.AttestationData{
-			Slot:            3600,
-			CommitteeIndex:  primitives.CommitteeIndex(i),
-			BeaconBlockRoot: bRoot,
-			Source: &ethpb.Checkpoint{
-				Epoch: primitives.Epoch(1000),
-				Root:  sRoot,
-			},
-			Target: &ethpb.Checkpoint{
-				Epoch: primitives.Epoch(2000),
-				Root:  tRoot,
-			},
-		}
-		attData = append(attData, ad)
-	}
-	allAtts := make([]*ethpb.Attestation, 0)
-
-	for _, ad := range attData {
-		copiedD := ad
-		for i := 0; i < 30; i++ {
-			att := &ethpb.Attestation{
-				AggregationBits: bitfield.NewBitlist(3),
-				Data:            copiedD,
-				Signature:       make([]byte, 96),
-			}
-			allAtts = append(allAtts, att)
-		}
-	}
-	return allAtts
-}
-
-func BenchmarkAttestationDataRoot(b *testing.B) {
-	allAtts := makeAtts(b)
-	b.ResetTimer()
-	b.ReportAllocs()
-
-	for i := 0; i < b.N; i++ {
-		for _, a := range allAtts {
-			_, err := a.Data.HashTreeRoot()
-			assert.NoError(b, err)
-		}
-	}
-}
-
-var lc = lruwrpr.New(100)
-
-func getRoot(attData *ethpb.AttestationData) ([32]byte, error) {
-	var key [128]byte
-
-	binary.LittleEndian.PutUint64(key[:8], uint64(attData.Slot))
-	binary.LittleEndian.PutUint64(key[8:16], uint64(attData.CommitteeIndex))
-	copy(key[16:48], attData.BeaconBlockRoot)
-	binary.LittleEndian.PutUint64(key[48:56], uint64(attData.Target.Epoch))
-	copy(key[56:88], attData.Target.Root)
-	binary.LittleEndian.PutUint64(key[88:96], uint64(attData.Source.Epoch))
-	copy(key[96:128], attData.Source.Root)
-
-	rt, ok := lc.Get(key)
-	if ok {
-		return rt.([32]byte), nil
-	}
-	htr, err := attData.HashTreeRoot()
-	if err != nil {
-		return [32]byte{}, err
-	}
-	lc.Add(key, htr)
-	return htr, nil
-}
-
-func BenchmarkAttestationDataRootWithCache(b *testing.B) {
-	allAtts := makeAtts(b)
-	b.ResetTimer()
-	b.ReportAllocs()
-
-	for i := 0; i < b.N; i++ {
-		for _, a := range allAtts {
-			rt, err := getRoot(a.Data)
-			_ = rt
-			assert.NoError(b, err)
-		}
-	}
-}
-
-var rtMap = map[uint64]map[uint64]map[[32]byte]map[uint64]map[[32]byte]map[uint64]map[[32]byte][32]byte{}
-var sltMap = map[uint64][]*ethpb.AttestationData{}
-var attMap = map[*ethpb.AttestationData][32]byte{}
-
-var mt = new(sync.RWMutex)
-
-func getRootInMap(attData *ethpb.AttestationData) ([32]byte, error) {
-	mt.RLock()
-	mpA, ok := rtMap[uint64(attData.Slot)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-	}
-	mt.RLock()
-	mpB, ok := mpA[uint64(attData.CommitteeIndex)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-
-	}
-	mt.RLock()
-	mpC, ok := mpB[[32]byte(attData.BeaconBlockRoot)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-
-	}
-	mt.RLock()
-	mpD, ok := mpC[uint64(attData.Target.Epoch)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-
-	}
-	mt.RLock()
-	mpE, ok := mpD[[32]byte(attData.Target.Root)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-
-	}
-	mt.RLock()
-	mpF, ok := mpE[uint64(attData.Target.Epoch)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-	}
-	mt.RLock()
-	htr, ok := mpF[[32]byte(attData.Target.Root)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMap(attData)
-	}
-	return htr, nil
-}
-
-func recoverRtInMap(attData *ethpb.AttestationData) ([32]byte, error) {
-	htr, err := attData.HashTreeRoot()
-	if err != nil {
-		return [32]byte{}, err
-	}
-	mt.Lock()
-	rtMap[uint64(attData.Slot)] = map[uint64]map[[32]byte]map[uint64]map[[32]byte]map[uint64]map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)] = map[[32]byte]map[uint64]map[[32]byte]map[uint64]map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)][[32]byte(attData.BeaconBlockRoot)] = map[uint64]map[[32]byte]map[uint64]map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)][[32]byte(attData.BeaconBlockRoot)][uint64(attData.Target.Epoch)] = map[[32]byte]map[uint64]map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)][[32]byte(attData.BeaconBlockRoot)][uint64(attData.Target.Epoch)][[32]byte(attData.Target.Root)] = map[uint64]map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)][[32]byte(attData.BeaconBlockRoot)][uint64(attData.Target.Epoch)][[32]byte(attData.Target.Root)][uint64(attData.Source.Epoch)] = map[[32]byte][32]byte{}
-	rtMap[uint64(attData.Slot)][uint64(attData.CommitteeIndex)][[32]byte(attData.BeaconBlockRoot)][uint64(attData.Target.Epoch)][[32]byte(attData.Target.Root)][uint64(attData.Source.Epoch)][[32]byte(attData.Source.Root)] = htr
-	mt.Unlock()
-	return htr, nil
-}
-
-func BenchmarkAttestationDataRootWithMap(b *testing.B) {
-	allAtts := makeAtts(b)
-	b.ResetTimer()
-	b.ReportAllocs()
-
-	for i := 0; i < b.N; i++ {
-		for _, a := range allAtts {
-			rt, err := getRootInMap(a.Data)
-			_ = rt
-			assert.NoError(b, err)
-		}
-	}
-}
-
-func recoverRtInMapNew(attData *ethpb.AttestationData) ([32]byte, error) {
-	htr, err := attData.HashTreeRoot()
-	if err != nil {
-		return [32]byte{}, err
-	}
-	mt.Lock()
-	sltMap[uint64(attData.Slot)] = append(sltMap[uint64(attData.Slot)], attData)
-	attMap[attData] = htr
-	mt.Unlock()
-	return htr, nil
-}
-
-func getRootInMapNew(attData *ethpb.AttestationData) ([32]byte, error) {
-	mt.RLock()
-	attList, ok := sltMap[uint64(attData.Slot)]
-	mt.RUnlock()
-	if !ok {
-		return recoverRtInMapNew(attData)
-	}
-	for _, a := range attList {
-		if attData.CommitteeIndex != a.CommitteeIndex {
-			continue
-		}
-		if !bytes.Equal(attData.BeaconBlockRoot, a.BeaconBlockRoot) {
-			continue
-		}
-		if attData.Target.Epoch != a.Target.Epoch {
-			continue
-		}
-		if !bytes.Equal(attData.Target.Root, a.Target.Root) {
-			continue
-		}
-		if attData.Source.Epoch != a.Source.Epoch {
-			continue
-		}
-		if !bytes.Equal(attData.Source.Root, a.Source.Root) {
-			continue
-		}
-		mt.RLock()
-		val, ok := attMap[a]
-		mt.RUnlock()
-		if !ok {
-			return recoverRtInMapNew(attData)
-		}
-		return val, nil
-	}
-	return recoverRtInMapNew(attData)
-}
-
-func BenchmarkAttestationDataRootWithMapNew(b *testing.B) {
-	allAtts := makeAtts(b)
-	b.ResetTimer()
-	b.ReportAllocs()
-
-	for i := 0; i < b.N; i++ {
-		for _, a := range allAtts {
-			rt, err := getRootInMapNew(a.Data)
-			_ = rt
-			assert.NoError(b, err)
-		}
-	}
-}

beacon-chain/db/filesystem/pruner.go

@@ -92,14 +92,16 @@ func windowMin(latest, offset primitives.Slot) primitives.Slot {
 
 func (p *blobPruner) warmCache() error {
 	p.Lock()
-	defer p.Unlock()
+	defer func() {
+		if !p.warmed {
+			p.warmed = true
+			close(p.cacheReady)
+		}
+		p.Unlock()
+	}()
 	if err := p.prune(0); err != nil {
 		return err
 	}
-	if !p.warmed {
-		p.warmed = true
-		close(p.cacheReady)
-	}
 	return nil
 }
 

beacon-chain/db/filesystem/pruner_test.go

@@ -2,16 +2,19 @@ package filesystem
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"math"
 	"os"
 	"path"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/verification"
 	fieldparams "github.com/prysmaticlabs/prysm/v5/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
 	"github.com/spf13/afero"
@@ -34,6 +37,34 @@ func TestTryPruneDir_CachedNotExpired(t *testing.T) {
 	require.Equal(t, 0, pruned)
 }
 
+func TestCacheWarmFail(t *testing.T) {
+	fs := afero.NewMemMapFs()
+	n := blobNamer{root: bytesutil.ToBytes32([]byte("derp")), index: 0}
+	bp := n.path()
+	mkdir := path.Dir(bp)
+	require.NoError(t, fs.MkdirAll(mkdir, directoryPermissions))
+
+	// Create an empty blob index in the fs by touching the file at a seemingly valid path.
+	fi, err := fs.Create(bp)
+	require.NoError(t, err)
+	require.NoError(t, fi.Close())
+
+	// Cache warm should fail due to the unexpected EOF.
+	pr, err := newBlobPruner(fs, 0)
+	require.NoError(t, err)
+	require.ErrorIs(t, pr.warmCache(), errPruningFailures)
+
+	// The cache warm has finished, so calling waitForCache with a super short deadline
+	// should not block or hit the context deadline.
+	ctx := context.Background()
+	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(1*time.Millisecond))
+	defer cancel()
+	c, err := pr.waitForCache(ctx)
+	// We will get an error and a nil value for the cache if we hit the deadline.
+	require.NoError(t, err)
+	require.NotNil(t, c)
+}
+
 func TestTryPruneDir_CachedExpired(t *testing.T) {
 	t.Run("empty directory", func(t *testing.T) {
 		fs := afero.NewMemMapFs()

beacon-chain/node/BUILD.bazel

@@ -16,7 +16,7 @@ go_library(
     ],
     deps = [
         "//api/gateway:go_default_library",
-        "//api/server:go_default_library",
+        "//api/server/middleware:go_default_library",
         "//async/event:go_default_library",
         "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/builder:go_default_library",

beacon-chain/node/node.go

@@ -20,7 +20,7 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
 	apigateway "github.com/prysmaticlabs/prysm/v5/api/gateway"
-	"github.com/prysmaticlabs/prysm/v5/api/server"
+	"github.com/prysmaticlabs/prysm/v5/api/server/middleware"
 	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/builder"
@@ -409,8 +409,8 @@ func newRouter(cliCtx *cli.Context) *mux.Router {
 		allowedOrigins = strings.Split(flags.GPRCGatewayCorsDomain.Value, ",")
 	}
 	r := mux.NewRouter()
-	r.Use(server.NormalizeQueryValuesHandler)
-	r.Use(server.CorsHandler(allowedOrigins))
+	r.Use(middleware.NormalizeQueryValuesHandler)
+	r.Use(middleware.CorsHandler(allowedOrigins))
 	return r
 }
 
@@ -965,9 +965,8 @@ func (b *BeaconNode) registerRPCService(router *mux.Router) error {
 	cert := b.cliCtx.String(flags.CertFlag.Name)
 	key := b.cliCtx.String(flags.KeyFlag.Name)
 	mockEth1DataVotes := b.cliCtx.Bool(flags.InteropMockEth1DataVotesFlag.Name)
-
 	maxMsgSize := b.cliCtx.Int(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
-	enableDebugRPCEndpoints := b.cliCtx.Bool(flags.EnableDebugRPCEndpoints.Name)
+	enableDebugRPCEndpoints := !b.cliCtx.Bool(flags.DisableDebugRPCEndpoints.Name)
 
 	p2pService := b.fetchP2P()
 	rpcService := rpc.NewService(b.ctx, &rpc.Config{
@@ -1056,11 +1055,10 @@ func (b *BeaconNode) registerGRPCGateway(router *mux.Router) error {
 	gatewayPort := b.cliCtx.Int(flags.GRPCGatewayPort.Name)
 	rpcHost := b.cliCtx.String(flags.RPCHost.Name)
 	rpcPort := b.cliCtx.Int(flags.RPCPort.Name)
-
+	enableDebugRPCEndpoints := !b.cliCtx.Bool(flags.DisableDebugRPCEndpoints.Name)
 	selfAddress := net.JoinHostPort(rpcHost, strconv.Itoa(rpcPort))
 	gatewayAddress := net.JoinHostPort(gatewayHost, strconv.Itoa(gatewayPort))
 	allowedOrigins := strings.Split(b.cliCtx.String(flags.GPRCGatewayCorsDomain.Name), ",")
-	enableDebugRPCEndpoints := b.cliCtx.Bool(flags.EnableDebugRPCEndpoints.Name)
 	selfCert := b.cliCtx.String(flags.CertFlag.Name)
 	maxCallSize := b.cliCtx.Uint64(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
 	httpModules := b.cliCtx.String(flags.HTTPModules.Name)

beacon-chain/rpc/BUILD.bazel

@@ -10,6 +10,8 @@ go_library(
     importpath = "github.com/prysmaticlabs/prysm/v5/beacon-chain/rpc",
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
+        "//api:go_default_library",
+        "//api/server/middleware:go_default_library",
         "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/builder:go_default_library",
         "//beacon-chain/cache:go_default_library",

beacon-chain/rpc/service.go

@@ -16,7 +16,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/builder"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/cache"
@@ -114,8 +113,8 @@ type Config struct {
 	ExecutionChainInfoFetcher     execution.ChainInfoFetcher
 	GenesisTimeFetcher            blockchain.TimeFetcher
 	GenesisFetcher                blockchain.GenesisFetcher
-	EnableDebugRPCEndpoints       bool
 	MockEth1Votes                 bool
+	EnableDebugRPCEndpoints       bool
 	AttestationsPool              attestations.Pool
 	ExitPool                      voluntaryexits.PoolManager
 	SlashingsPool                 slashings.PoolManager
@@ -311,13 +310,7 @@ func NewService(ctx context.Context, cfg *Config) *Service {
 	for _, e := range endpoints {
 		s.cfg.Router.HandleFunc(
 			e.template,
-			promhttp.InstrumentHandlerDuration(
-				httpRequestLatency.MustCurryWith(prometheus.Labels{"endpoint": e.name}),
-				promhttp.InstrumentHandlerCounter(
-					httpRequestCount.MustCurryWith(prometheus.Labels{"endpoint": e.name}),
-					e.handler,
-				),
-			),
+			e.handlerWithMiddleware(),
 		).Methods(e.methods...)
 	}
 
@@ -325,7 +318,6 @@ func NewService(ctx context.Context, cfg *Config) *Service {
 	ethpbv1alpha1.RegisterHealthServer(s.grpcServer, nodeServer)
 	ethpbv1alpha1.RegisterBeaconChainServer(s.grpcServer, beaconChainServer)
 	if s.cfg.EnableDebugRPCEndpoints {
-		log.Info("Enabled debug gRPC endpoints")
 		debugServer := &debugv1alpha1.Server{
 			GenesisTimeFetcher: s.cfg.GenesisTimeFetcher,
 			BeaconDB:           s.cfg.BeaconDB,

beacon-chain/sync/validate_beacon_blocks.go

@@ -3,6 +3,8 @@ package sync
 import (
 	"context"
 	"fmt"
+	"os"
+	"runtime/pprof"
 	"time"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -35,10 +37,47 @@ var (
 	errRejectCommitmentLen = errors.New("[REJECT] The length of KZG commitments is less than or equal to the limitation defined in Consensus Layer")
 )
 
+func (s *Service) validateBeaconBlockPubSub(ctx context.Context, pid peer.ID, msg *pubsub.Message) (pubsub.ValidationResult, error) {
+	done := make(chan struct {
+		result pubsub.ValidationResult
+		err    error
+	})
+
+	go func() {
+		result, err := func() (pubsub.ValidationResult, error) {
+			return s.validateBeaconBlockPubSubInternal(ctx, pid, msg)
+		}()
+		done <- struct {
+			result pubsub.ValidationResult
+			err    error
+		}{result, err}
+	}()
+
+	select {
+	case res := <-done:
+		return res.result, res.err
+	case <-time.After(12 * time.Second):
+		filePath := "/tmp/pre-confirm-blocks.goroutine"
+		file, err := os.Create(filePath)
+		if err != nil {
+			log.WithError(err).Error("Could not create file for goroutine dump")
+		}
+		defer func() {
+			if err := file.Close(); err != nil {
+				log.WithError(err).Error("Could not close file for goroutine dump")
+			}
+		}()
+		if err := pprof.Lookup("goroutine").WriteTo(file, 2); err != nil {
+			log.WithError(err).Error("Could not write goroutine dump to file")
+		}
+	}
+	return pubsub.ValidationIgnore, nil
+}
+
 // validateBeaconBlockPubSub checks that the incoming block has a valid BLS signature.
 // Blocks that have already been seen are ignored. If the BLS signature is any valid signature,
 // this method rebroadcasts the message.
-func (s *Service) validateBeaconBlockPubSub(ctx context.Context, pid peer.ID, msg *pubsub.Message) (pubsub.ValidationResult, error) {
+func (s *Service) validateBeaconBlockPubSubInternal(ctx context.Context, pid peer.ID, msg *pubsub.Message) (pubsub.ValidationResult, error) {
 	receivedTime := prysmTime.Now()
 	// Validation runs on publish (not just subscriptions), so we should approve any message from
 	// ourselves.

cmd/beacon-chain/flags/base.go

@@ -174,10 +174,10 @@ var (
 		Usage: "The factor by which blob batch limit may increase on burst.",
 		Value: 2,
 	}
-	// EnableDebugRPCEndpoints as /v1/beacon/state.
-	EnableDebugRPCEndpoints = &cli.BoolFlag{
-		Name:  "enable-debug-rpc-endpoints",
-		Usage: "Enables the debug rpc service, containing utility endpoints such as /eth/v1alpha1/beacon/state.",
+	// DisableDebugRPCEndpoints disables the debug Beacon API namespace.
+	DisableDebugRPCEndpoints = &cli.BoolFlag{
+		Name:  "disable-debug-rpc-endpoints",
+		Usage: "Disables the debug Beacon API namespace.",
 	}
 	// SubscribeToAllSubnets defines a flag to specify whether to subscribe to all possible attestation/sync subnets or not.
 	SubscribeToAllSubnets = &cli.BoolFlag{

cmd/beacon-chain/main.go

@@ -64,7 +64,7 @@ var appFlags = []cli.Flag{
 	flags.InteropNumValidatorsFlag,
 	flags.InteropGenesisTimeFlag,
 	flags.SlotsPerArchivedPoint,
-	flags.EnableDebugRPCEndpoints,
+	flags.DisableDebugRPCEndpoints,
 	flags.SubscribeToAllSubnets,
 	flags.HistoricalSlasherNode,
 	flags.ChainID,

cmd/beacon-chain/usage.go

@@ -116,7 +116,7 @@ var appHelpFlagGroups = []flagGroup{
 			flags.BlockBatchLimitBurstFactor,
 			flags.BlobBatchLimit,
 			flags.BlobBatchLimitBurstFactor,
-			flags.EnableDebugRPCEndpoints,
+			flags.DisableDebugRPCEndpoints,
 			flags.SubscribeToAllSubnets,
 			flags.HistoricalSlasherNode,
 			flags.ChainID,

config/features/deprecated_flags.go

@@ -52,6 +52,11 @@ var (
 		Usage:  deprecatedUsage,
 		Hidden: true,
 	}
+	deprecatedEnableDebugRPCEndpoints = &cli.BoolFlag{
+		Name:   "enable-debug-rpc-endpoints",
+		Usage:  deprecatedUsage,
+		Hidden: true,
+	}
 )
 
 // Deprecated flags for both the beacon node and validator client.
@@ -65,6 +70,7 @@ var deprecatedFlags = []cli.Flag{
 	deprecatedEnableEIP4881,
 	deprecatedDisableEIP4881,
 	deprecatedVerboseSigVerification,
+	deprecatedEnableDebugRPCEndpoints,
 }
 
 // deprecatedBeaconFlags contains flags that are still used by other components

testing/endtoend/components/beacon_node.go

@@ -275,7 +275,6 @@ func (node *BeaconNode) Start(ctx context.Context) error {
 		"--" + cmdshared.ValidatorMonitorIndicesFlag.Name + "=2",
 		"--" + cmdshared.ForceClearDB.Name,
 		"--" + cmdshared.AcceptTosFlag.Name,
-		"--" + flags.EnableDebugRPCEndpoints.Name,
 		"--" + features.EnableQUIC.Name,
 	}
 	if config.UsePprof {

validator/node/BUILD.bazel

@@ -37,7 +37,7 @@ go_library(
     deps = [
         "//api:go_default_library",
         "//api/gateway:go_default_library",
-        "//api/server:go_default_library",
+        "//api/server/middleware:go_default_library",
         "//async/event:go_default_library",
         "//cmd:go_default_library",
         "//cmd/validator/flags:go_default_library",

validator/node/node.go

@@ -25,7 +25,7 @@ import (
 	fastssz "github.com/prysmaticlabs/fastssz"
 	"github.com/prysmaticlabs/prysm/v5/api"
 	"github.com/prysmaticlabs/prysm/v5/api/gateway"
-	"github.com/prysmaticlabs/prysm/v5/api/server"
+	"github.com/prysmaticlabs/prysm/v5/api/server/middleware"
 	"github.com/prysmaticlabs/prysm/v5/async/event"
 	"github.com/prysmaticlabs/prysm/v5/cmd"
 	"github.com/prysmaticlabs/prysm/v5/cmd/validator/flags"
@@ -155,8 +155,8 @@ func newRouter(cliCtx *cli.Context) *mux.Router {
 		allowedOrigins = strings.Split(flags.GRPCGatewayCorsDomain.Value, ",")
 	}
 	r := mux.NewRouter()
-	r.Use(server.NormalizeQueryValuesHandler)
-	r.Use(server.CorsHandler(allowedOrigins))
+	r.Use(middleware.NormalizeQueryValuesHandler)
+	r.Use(middleware.CorsHandler(allowedOrigins))
 	return r
 }