add replication outbound msgs

Author: samliok

api/grpcclient/client.go

@@ -0,0 +1,71 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package grpcclient
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/ava-labs/avalanchego/ids"
+)
+
+// NewChainClient returns a grpc.ClientConn that sets the chain-id header for
+// all requests
+func NewChainClient(uri string, chainID ids.ID, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+	dialOpts := []grpc.DialOption{
+		grpc.WithUnaryInterceptor(SetChainIDHeaderUnaryClientInterceptor(chainID)),
+		grpc.WithStreamInterceptor(SetChainIDHeaderStreamClientInterceptor(chainID)),
+	}
+
+	dialOpts = append(dialOpts, opts...)
+
+	conn, err := grpc.NewClient(uri, dialOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize chain grpc client: %w", err)
+	}
+
+	return conn, nil
+}
+
+// SetChainIDHeaderUnaryClientInterceptor sets the chain-id header for unary
+// requests
+func SetChainIDHeaderUnaryClientInterceptor(chainID ids.ID) grpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		req any,
+		reply any,
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		opts ...grpc.CallOption,
+	) error {
+		ctx = newContextWithChainIDHeader(ctx, chainID)
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+// SetChainIDHeaderStreamClientInterceptor sets the chain-id header for
+// streaming requests
+func SetChainIDHeaderStreamClientInterceptor(chainID ids.ID) grpc.StreamClientInterceptor {
+	return func(
+		ctx context.Context,
+		desc *grpc.StreamDesc,
+		cc *grpc.ClientConn,
+		method string,
+		streamer grpc.Streamer,
+		opts ...grpc.CallOption,
+	) (grpc.ClientStream, error) {
+		ctx = newContextWithChainIDHeader(ctx, chainID)
+		return streamer(ctx, desc, cc, method, opts...)
+	}
+}
+
+// newContextWithChainHeader sets the chain-id header which the server uses
+// to route the client grpc request
+func newContextWithChainIDHeader(ctx context.Context, chainID ids.ID) context.Context {
+	return metadata.AppendToOutgoingContext(ctx, "chain-id", chainID.String())
+}

api/server/http2_router.go

@@ -0,0 +1,57 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package server
+
+import (
+	"net/http"
+	"sync"
+
+	"github.com/ava-labs/avalanchego/ids"
+)
+
+var _ http.Handler = (*http2Router)(nil)
+
+type http2Router struct {
+	lock     sync.RWMutex
+	handlers map[string]http.Handler
+}
+
+func newHTTP2Router() *http2Router {
+	return &http2Router{
+		handlers: make(map[string]http.Handler),
+	}
+}
+
+func (h *http2Router) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// the chain-id header must be set to route the request to the correct chain
+	// http2 handler
+	chainID := r.Header.Get("chain-id")
+	if len(chainID) == 0 {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	h.lock.RLock()
+	handler, ok := h.handlers[chainID]
+	h.lock.RUnlock()
+	if !ok {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	handler.ServeHTTP(w, r)
+}
+
+func (h *http2Router) Add(chainID ids.ID, handler http.Handler) bool {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	chainIDStr := chainID.String()
+	if _, ok := h.handlers[chainIDStr]; ok {
+		return false
+	}
+
+	h.handlers[chainIDStr] = handler
+	return true
+}

api/server/http2_router_test.go

@@ -0,0 +1,71 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package server
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ava-labs/avalanchego/ids"
+)
+
+func TestHTTP2RouterAdd(t *testing.T) {
+	require := require.New(t)
+	h := newHTTP2Router()
+	handler := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})
+
+	require.True(h.Add(ids.Empty, handler))
+	require.False(h.Add(ids.Empty, handler))
+}
+
+func TestHTTP2RouterServeHTTP(t *testing.T) {
+	tests := []struct {
+		name     string
+		chainIDs []ids.ID
+		header   http.Header
+		wantCode int
+	}{
+		{
+			name:     "missing chain-id header",
+			wantCode: http.StatusBadRequest,
+		},
+		{
+			name: "unknown referenced chain-id",
+			header: http.Header{
+				http.CanonicalHeaderKey("chain-id"): []string{ids.Empty.String()},
+			},
+			wantCode: http.StatusNotFound,
+		},
+		{
+			name:     "valid handler",
+			chainIDs: []ids.ID{ids.Empty},
+			header: http.Header{
+				http.CanonicalHeaderKey("chain-id"): []string{ids.Empty.String()},
+			},
+			wantCode: http.StatusOK,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require := require.New(t)
+
+			h := newHTTP2Router()
+			handler := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})
+			writer := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodPost, "/", nil)
+			request.Header = tt.header
+
+			for _, chainID := range tt.chainIDs {
+				require.True(h.Add(chainID, handler))
+			}
+
+			h.ServeHTTP(writer, request)
+			require.Equal(tt.wantCode, writer.Code)
+		})
+	}
+}

api/server/server.go

@@ -12,11 +12,11 @@ import (
 	"path"
 	"time"
 
-	"github.com/NYTimes/gziphandler"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/rs/cors"
 	"go.uber.org/zap"
 	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
 
 	"github.com/ava-labs/avalanchego/api"
 	"github.com/ava-labs/avalanchego/ids"
@@ -88,7 +88,8 @@ type server struct {
 	metrics *metrics
 
 	// Maps endpoints to handlers
-	router *router
+	router      *router
+	http2Router *http2Router
 
 	srv *http.Server
 
@@ -115,33 +116,29 @@ func New(
 	}
 
 	router := newRouter()
-	allowedHostsHandler := filterInvalidHosts(router, allowedHosts)
-	corsHandler := cors.New(cors.Options{
-		AllowedOrigins:   allowedOrigins,
-		AllowCredentials: true,
-	}).Handler(allowedHostsHandler)
-	gzipHandler := gziphandler.GzipHandler(corsHandler)
-	var handler http.Handler = http.HandlerFunc(
-		func(w http.ResponseWriter, r *http.Request) {
-			// Attach this node's ID as a header
-			w.Header().Set("node-id", nodeID.String())
-			gzipHandler.ServeHTTP(w, r)
-		},
-	)
+	handler := wrapHandler(router, nodeID, allowedOrigins, allowedHosts)
+
+	http2Router := newHTTP2Router()
+	http2Handler := wrapHandler(http2Router, nodeID, allowedOrigins, allowedHosts)
 
 	httpServer := &http.Server{
-		Handler:           handler,
+		Handler: h2c.NewHandler(
+			http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.ProtoMajor == 2 {
+					http2Handler.ServeHTTP(w, r)
+					return
+				}
+
+				handler.ServeHTTP(w, r)
+			}),
+			&http2.Server{
+				MaxConcurrentStreams: maxConcurrentStreams,
+			}),
 		ReadTimeout:       httpConfig.ReadTimeout,
 		ReadHeaderTimeout: httpConfig.ReadHeaderTimeout,
 		WriteTimeout:      httpConfig.WriteTimeout,
 		IdleTimeout:       httpConfig.IdleTimeout,
 	}
-	err = http2.ConfigureServer(httpServer, &http2.Server{
-		MaxConcurrentStreams: maxConcurrentStreams,
-	})
-	if err != nil {
-		return nil, err
-	}
 
 	log.Info("API created",
 		zap.Strings("allowedOrigins", allowedOrigins),
@@ -154,6 +151,7 @@ func New(
 		tracer:          tracer,
 		metrics:         m,
 		router:          router,
+		http2Router:     http2Router,
 		srv:             httpServer,
 		listener:        listener,
 	}, nil
@@ -199,6 +197,30 @@ func (s *server) RegisterChain(chainName string, ctx *snow.ConsensusContext, vm
 			)
 		}
 	}
+
+	ctx.Lock.Lock()
+	http2Handler, err := vm.CreateHTTP2Handler(context.TODO())
+	ctx.Lock.Unlock()
+	if err != nil {
+		s.log.Error("failed to create http2 handler",
+			zap.String("chainName", chainName),
+			zap.Error(err),
+		)
+		return
+	}
+
+	if http2Handler == nil {
+		return
+	}
+
+	http2Handler = s.wrapMiddleware(chainName, http2Handler, ctx)
+	if !s.http2Router.Add(ctx.ChainID, http2Handler) {
+		s.log.Error(
+			"failed to add route to http2 handler",
+			zap.String("chainName", chainName),
+			zap.Error(err),
+		)
+	}
 }
 
 func (s *server) addChainRoute(chainName string, handler http.Handler, ctx *snow.ConsensusContext, base, endpoint string) error {
@@ -207,13 +229,17 @@ func (s *server) addChainRoute(chainName string, handler http.Handler, ctx *snow
 		zap.String("url", url),
 		zap.String("endpoint", endpoint),
 	)
+	handler = s.wrapMiddleware(chainName, handler, ctx)
+	return s.router.AddRouter(url, endpoint, handler)
+}
+
+func (s *server) wrapMiddleware(chainName string, handler http.Handler, ctx *snow.ConsensusContext) http.Handler {
 	if s.tracingEnabled {
 		handler = api.TraceHandler(handler, chainName, s.tracer)
 	}
 	// Apply middleware to reject calls to the handler before the chain finishes bootstrapping
 	handler = rejectMiddleware(handler, ctx)
-	handler = s.metrics.wrapHandler(chainName, handler)
-	return s.router.AddRouter(url, endpoint, handler)
+	return s.metrics.wrapHandler(chainName, handler)
 }
 
 func (s *server) AddRoute(handler http.Handler, base, endpoint string) error {
@@ -299,3 +325,23 @@ func (a readPathAdder) AddRoute(handler http.Handler, base, endpoint string) err
 func (a readPathAdder) AddAliases(endpoint string, aliases ...string) error {
 	return a.pather.AddAliasesWithReadLock(endpoint, aliases...)
 }
+
+func wrapHandler(
+	handler http.Handler,
+	nodeID ids.NodeID,
+	allowedOrigins []string,
+	allowedHosts []string,
+) http.Handler {
+	h := filterInvalidHosts(handler, allowedHosts)
+	h = cors.New(cors.Options{
+		AllowedOrigins:   allowedOrigins,
+		AllowCredentials: true,
+	}).Handler(h)
+	return http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			// Attach this node's ID as a header
+			w.Header().Set("node-id", nodeID.String())
+			h.ServeHTTP(w, r)
+		},
+	)
+}

go.mod

@@ -11,12 +11,11 @@ go 1.23.9
 
 require (
 	github.com/DataDog/zstd v1.5.2
-	github.com/NYTimes/gziphandler v1.1.1
 	github.com/StephenButtolph/canoto v0.15.0
 	github.com/antithesishq/antithesis-sdk-go v0.3.8
-	github.com/ava-labs/coreth v0.15.1-rc.0.0.20250530184801-28421010abae
+	github.com/ava-labs/coreth v0.15.2-rc.0.0.20250610170140-2fcf45f828a2
 	github.com/ava-labs/ledger-avalanche/go v0.0.0-20241009183145-e6f90a8a1a60
-	github.com/ava-labs/libevm v1.13.14-0.2.0.release
+	github.com/ava-labs/libevm v0.0.0-20250610142802-2672fbd7cdfc
 	github.com/btcsuite/btcd/btcutil v1.1.3
 	github.com/cockroachdb/pebble v0.0.0-20230928194634-aa077af62593
 	github.com/compose-spec/compose-go v1.20.2