core/validatorapi: support proxy of SSE events

app/eth2wrap/httpwrap.go

@@ -135,3 +135,7 @@ func (h *httpAdapter) Proxy(ctx context.Context, req *http.Request) (*http.Respo
 	log.Debug(ctx, "Proxying request to beacon node", z.Any("url", h.address))
 	return h.Service.Proxy(ctx, req)
 }
+
+func (h *httpAdapter) Headers() map[string]string {
+	return h.headers
+}

app/eth2wrap/lazy.go

@@ -120,6 +120,15 @@ func (l *lazy) Address() string {
 	return cl.Address()
 }
 
+func (l *lazy) Headers() map[string]string {
+	cl, ok := l.getClient()
+	if !ok {
+		return nil
+	}
+
+	return cl.Headers()
+}
+
 func (l *lazy) IsActive() bool {
 	cl, ok := l.getClient()
 	if !ok {

app/eth2wrap/multi.go

@@ -59,6 +59,13 @@ func (m multi) Address() string {
 	return address
 }
 
+func (m multi) Headers() map[string]string {
+	if len(m.clients) == 0 {
+		return nil
+	}
+	return m.clients[0].Headers()
+}
+
 func (m multi) IsActive() bool {
 	for _, cl := range m.clients {
 		if cl.IsActive() {

core/validatorapi/router.go

@@ -11,9 +11,11 @@
 	"encoding/json"
 	"fmt"
 	"io"
+	stdlog "log"
 	"maps"
 	"math"
 	"net/http"
+	"net/http/httputil"
 	"net/url"
 	"slices"
 	"strconv"
@@ -83,6 +85,11 @@
 	eth2client.ValidatorRegistrationsSubmitter
 	eth2client.VoluntaryExitSubmitter
 	// Above sorted alphabetically.
+
+	// Address returns the address of the beacon node.
+	Address() string
+	// Headers returns custom headers to include in requests to the beacon node.
+	Headers() map[string]string
 }
 
 // NewRouter returns a new validator http server router. The http router
@@ -317,6 +324,9 @@
 		}
 	}
 
+	// SSE events endpoint requires raw HTTP access for streaming
+	r.Handle("/eth/v1/events", eventsHandler(h)).Methods(http.MethodGet)
+
 	// Everything else is proxied
 	r.PathPrefix("/").Handler(proxy(h))
 
@@ -1599,12 +1609,86 @@
 	}
 }
 
-func proxy(p eth2client.ProxyProvider) http.HandlerFunc {
+// writeFlusher is copied from /net/http/httputil/reverseproxy.go.
+// It is required to flush streaming responses.
+type writeFlusher interface {
+	http.ResponseWriter
+	http.Flusher
+}
+
+// proxyResponseWriter wraps the writeFlusher interface and instruments errors.
+type proxyResponseWriter struct {
+	writeFlusher
+}
+
+func (w proxyResponseWriter) WriteHeader(statusCode int) {
+	if statusCode/100 == 2 {
+		// 2XX isn't an error
+		return
+	}
+
+	incAPIErrors("proxy", statusCode)
+	w.writeFlusher.WriteHeader(statusCode)
+}
+
+// eventsHandler directly reverse proxies the SSE request to one of the configured beacon nodes.
+// This is used for SSE endpoints which require a persistent connection.
+func eventsHandler(h Handler) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		ctx = log.WithTopic(ctx, "vapi")
+		ctx = log.WithCtx(ctx, z.Str("vapi_events_method", r.Method), z.Str("vapi_events_path", r.URL.Path))
+		ctx = withCtxDuration(ctx)
+
+		// Get one of the configured beacon node addresses for proxying.
+		beaconNodeAddr := h.Address()
+		headers := h.Headers()
+
+		targetURL, err := url.ParseRequestURI(beaconNodeAddr)
+		if err != nil {
+			log.Error(ctx, "Failed to parse beacon node address for proxying", err, z.Str("address", beaconNodeAddr))
+			writeError(ctx, w, "events", err)
+			return
+		}
+
+		proxy := httputil.NewSingleHostReverseProxy(targetURL)
+
+		// Extend default proxy director with basic auth and host header
+		defaultDirector := proxy.Director
+		proxy.Director = func(req *http.Request) {
+			if targetURL.User != nil {
+				password, _ := targetURL.User.Password()
+				req.SetBasicAuth(targetURL.User.Username(), password)
+			}
+
+			req.Host = targetURL.Host
+			defaultDirector(req)
+
+			// Apply user provided beacon node headers
+			for k, v := range headers {
+				req.Header.Set(k, v)
+			}
+		}
+
+		proxy.ErrorLog = stdlog.New(io.Discard, "", 0)
+
+		// Use provided context for proxied requests, so long running
+		// requests are cancelled when this context is cancelled (soft shutdown).
+		clonedReq := r.Clone(ctx)
+
+		log.Debug(ctx, "Reverse proxying SSE request to beacon node", z.Str("beacon_node_address", beaconNodeAddr))
+
+		proxy.ServeHTTP(proxyResponseWriter{w.(writeFlusher)}, clonedReq)
+	}
+}
+
+func proxy(h Handler) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		ctx = log.WithTopic(ctx, "vapi")
 		ctx = log.WithCtx(ctx, z.Str("vapi_proxy_method", r.Method), z.Str("vapi_proxy_path", r.URL.Path))
 		ctx = withCtxDuration(ctx)
+
 		ctx, cancel := context.WithTimeout(ctx, defaultRequestTimeout)
 
 		defer func() {
@@ -1616,7 +1700,7 @@
 			cancel()
 		}()
 
-		res, err := p.Proxy(ctx, r)
+		res, err := h.Proxy(ctx, r)
 		if err != nil {
 			writeError(ctx, w, r.URL.Path, err)
 			return

core/validatorapi/router_internal_test.go

@@ -13,6 +13,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
+	"os"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -48,6 +49,80 @@ const (
 	infoLevel        = 1 // 1 is InfoLevel, this avoids importing zerolog directly.
 )
 
+func TestProxyShutdown(t *testing.T) {
+	// Start a server that will block until the request is cancelled.
+	serving := make(chan struct{})
+	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		close(serving)
+		<-r.Context().Done()
+	}))
+	defer target.Close()
+
+	// Create a handler that proxies to the target server
+	handler := testHandler{
+		AddressFunc: func() string { return target.URL },
+		ProxyFunc: func(ctx context.Context, req *http.Request) (*http.Response, error) {
+			// Forward the request to the target server
+			proxyReq, err := http.NewRequestWithContext(ctx, req.Method, target.URL+req.URL.Path, req.Body)
+			if err != nil {
+				return nil, errors.Wrap(err, "create proxy request")
+			}
+			proxyReq.Header = req.Header
+
+			client := &http.Client{}
+			return client.Do(proxyReq)
+		},
+	}
+
+	// Start a proxy server that will proxy to the target server.
+	ctx, cancel := context.WithCancel(t.Context())
+	proxyHTTP := proxy(handler)
+	proxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		proxyHTTP.ServeHTTP(w, r.WithContext(ctx))
+	}))
+	defer proxyServer.Close()
+
+	// Make a request to the proxy server, this will block until the proxy is shutdown.
+	errCh := make(chan error, 1)
+
+	go func() {
+		_, err := http.Get(proxyServer.URL)
+		errCh <- err
+	}()
+
+	// Wait for the target server is serving the request.
+	<-serving
+	// Shutdown the proxy server.
+	cancel()
+	// Wait for the request to complete.
+	err := <-errCh
+	require.NoError(t, err)
+}
+
+func TestRouterIntegration(t *testing.T) {
+	beaconURL, ok := os.LookupEnv("BEACON_URL")
+	if !ok {
+		t.Skip("Skipping integration test since BEACON_URL not found")
+	}
+
+	handler := testHandler{}
+	// Override Address to return the beacon URL from environment
+	handler.AddressFunc = func() string {
+		return beaconURL
+	}
+
+	r, err := NewRouter(handler, true)
+	require.NoError(t, err)
+
+	server := httptest.NewServer(r)
+	defer server.Close()
+
+	resp, err := http.Get(server.URL + "/eth/v1/node/version")
+	require.NoError(t, err)
+
+	require.Equal(t, 200, resp.StatusCode)
+}
+
 func TestRawRouter(t *testing.T) {
 	t.Run("proxy", func(t *testing.T) {
 		handler := testHandler{
@@ -2097,6 +2172,8 @@ type testHandler struct {
 	SyncCommitteeDutiesFunc          func(ctx context.Context, opts *eth2api.SyncCommitteeDutiesOpts) (*eth2api.Response[[]*eth2v1.SyncCommitteeDuty], error)
 	SyncCommitteeContributionFunc    func(ctx context.Context, opts *eth2api.SyncCommitteeContributionOpts) (*eth2api.Response[*altair.SyncCommitteeContribution], error)
 	ProxyFunc                        func(ctx context.Context, req *http.Request) (*http.Response, error)
+	AddressFunc                      func() string
+	HeadersFunc                      func() map[string]string
 }
 
 func (h testHandler) AttestationData(ctx context.Context, opts *eth2api.AttestationDataOpts) (*eth2api.Response[*eth2p0.AttestationData], error) {
@@ -2191,6 +2268,22 @@ func (h testHandler) Proxy(ctx context.Context, req *http.Request) (*http.Respon
 	return h.ProxyFunc(ctx, req)
 }
 
+func (h testHandler) Address() string {
+	if h.AddressFunc != nil {
+		return h.AddressFunc()
+	}
+
+	return "http://mock-beacon-node"
+}
+
+func (h testHandler) Headers() map[string]string {
+	if h.HeadersFunc != nil {
+		return h.HeadersFunc()
+	}
+
+	return nil // Test handler doesn't use custom headers by default
+}
+
 // newBeaconHandler returns a mock beacon node handler. It registers a few mock handlers required by the
 // eth2http service on startup, all other requests are routed to ProxyHandler if not nil.
 func (h testHandler) newBeaconHandler(t *testing.T) http.Handler {

core/validatorapi/validatorapi.go

@@ -1304,6 +1304,14 @@ func (c Component) Proxy(ctx context.Context, req *http.Request) (*http.Response
 	return c.eth2Cl.Proxy(ctx, req)
 }
 
+func (c Component) Address() string {
+	return c.eth2Cl.Address()
+}
+
+func (c Component) Headers() map[string]string {
+	return c.eth2Cl.Headers()
+}
+
 // wrapResponse wraps the provided data into an API Response and returns the response.
 func wrapResponse[T any](data T) *eth2api.Response[T] {
 	return &eth2api.Response[T]{Data: data}

testutil/beaconmock/beaconmock.go

@@ -463,6 +463,10 @@ func (m Mock) Address() string {
 	return "http://" + m.httpServer.Addr
 }
 
+func (Mock) Headers() map[string]string {
+	return nil // Mock doesn't use custom headers
+}
+
 func (m Mock) IsActive() bool {
 	return m.IsActiveFunc()
 }