refactor(rpcprovider)!: use pure gRPC instead of grpc-web wrapper

BREAKING CHANGE: gRPC-web protocol is no longer supported. Browser-based
clients that relied on grpc-web to communicate directly with providers
will no longer work. Only native gRPC clients are supported.

The previous implementation used grpc-web and h2c HTTP wrappers to support
the Lava SDK (lava-sdk/lavajs) which allowed browser-based JavaScript clients
to communicate with providers. Since SDK support has been removed (see #2186),
this infrastructure is no longer needed.

Replace the grpc-web/h2c/HTTP hybrid server with a cleaner architecture:

- Use cmux to multiplex HTTP health checks and native gRPC on same port
- Remove grpcweb.WrapServer and h2c.NewHandler complexity
- Add gRPC health checking protocol support (grpc_health_v1)
- Keep HTTP health endpoint for Kubernetes probes compatibility
- Optimize cmux matching: HTTP1Fast() first, then Any() for gRPC
  (avoids expensive HTTP/2 header parsing at high scale)

Benefits:
- Native gRPC connection management (keepalives, stream limits)
- Proper graceful shutdown via grpcServer.GracefulStop()
- Simpler, more maintainable code
- Better performance at scale (no protocol translation overhead)

Also adds comprehensive test suite for provider_listener.go with
14 tests and 2 benchmarks covering HTTP health, gRPC health,
relay/probe RPCs, concurrent requests, and error handling.

Author: nimrod-teich

go.mod

@@ -134,6 +134,7 @@ require (
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
+	github.com/soheilhy/cmux v0.1.5 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/supranational/blst v0.3.13 // indirect
 	github.com/tidwall/btree v1.6.0 // indirect

go.sum

@@ -1218,6 +1218,8 @@ github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVs
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
+github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
+github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
 github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
 github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
@@ -1488,6 +1490,7 @@ golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201031054903-ff519b6c9102/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210220033124-5f55cee0dc0d/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=

protocol/rpcprovider/provider_listener.go

@@ -2,35 +2,52 @@ package rpcprovider
 
 import (
 	"context"
-	"errors"
-	"fmt"
+	"crypto/tls"
+	"net"
 	"net/http"
 	"strings"
 	"sync"
 
 	"github.com/gogo/status"
-	"github.com/improbable-eng/grpc-web/go/grpcweb"
 	"github.com/lavanet/lava/v5/protocol/chainlib"
-	"github.com/lavanet/lava/v5/protocol/common"
 	"github.com/lavanet/lava/v5/protocol/lavaprotocol/protocolerrors"
 	"github.com/lavanet/lava/v5/protocol/lavasession"
 	"github.com/lavanet/lava/v5/utils"
 	pairingtypes "github.com/lavanet/lava/v5/x/pairing/types"
-	"golang.org/x/net/http2"
-	"golang.org/x/net/http2/h2c"
+	"github.com/soheilhy/cmux"
 	grpc "google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health"
+	healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
 )
 
 const (
 	HealthCheckURLPathFlagName    = "health-check-url-path"
 	HealthCheckURLPathFlagDefault = "/lava/health"
 )
 
+// isShutdownError returns true if the error is expected during graceful shutdown
+func isShutdownError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if err == http.ErrServerClosed || err == cmux.ErrListenerClosed || err == net.ErrClosed {
+		return true
+	}
+	// Check error message for common shutdown patterns
+	errStr := err.Error()
+	return strings.Contains(errStr, "use of closed network connection") ||
+		strings.Contains(errStr, "server closed") ||
+		strings.Contains(errStr, "mux: listener closed")
+}
+
 type ProviderListener struct {
 	networkAddress string
 	relayServer    *relayServer
-	httpServer     http.Server
+	grpcServer     *grpc.Server
+	httpServer     *http.Server
+	healthServer   *health.Server
+	cmux           cmux.CMux
 }
 
 func (pl *ProviderListener) Key() string {
@@ -47,69 +64,104 @@ func (pl *ProviderListener) RegisterReceiver(existingReceiver RelayReceiver, end
 		return utils.LavaFormatError("double_receiver_setup receiver already defined on this address with the same chainID and apiInterface", nil, utils.Attribute{Key: "chainID", Value: endpoint.ChainID}, utils.Attribute{Key: "apiInterface", Value: endpoint.ApiInterface})
 	}
 	pl.relayServer.relayReceivers[listen_endpoint.Key()] = &relayReceiverWrapper{relayReceiver: &existingReceiver, enabled: true}
+	// Mark service as healthy when receiver is registered
+	serviceName := endpoint.ChainID + "-" + endpoint.ApiInterface
+	pl.healthServer.SetServingStatus(serviceName, healthgrpc.HealthCheckResponse_SERVING)
 	utils.LavaFormatInfo("[++] Provider Listening on Address", utils.Attribute{Key: "chainID", Value: endpoint.ChainID}, utils.Attribute{Key: "apiInterface", Value: endpoint.ApiInterface}, utils.Attribute{Key: "Address", Value: endpoint.NetworkAddress})
 	return nil
 }
 
 func (pl *ProviderListener) Shutdown(shutdownCtx context.Context) error {
-	if err := pl.httpServer.Shutdown(shutdownCtx); err != nil {
-		utils.LavaFormatFatal("Provider failed to shutdown", err)
-	}
+	pl.healthServer.Shutdown()
+	pl.httpServer.Shutdown(shutdownCtx)
+	pl.grpcServer.GracefulStop()
 	return nil
 }
 
 func NewProviderListener(ctx context.Context, networkAddress lavasession.NetworkAddressData, healthCheckPath string) *ProviderListener {
 	pl := &ProviderListener{networkAddress: networkAddress.Address}
 
-	// GRPC
 	lis := chainlib.GetListenerWithRetryGrpc("tcp", networkAddress.Address)
-	opts := []grpc.ServerOption{
-		grpc.MaxRecvMsgSize(1024 * 1024 * 512), // setting receive size to 512mb for large debug responses
-		grpc.MaxSendMsgSize(1024 * 1024 * 512), // setting send size to 512mb for large debug responses
-	}
-	grpcServer := grpc.NewServer(opts...)
-	wrappedServer := grpcweb.WrapServer(grpcServer)
-	handler := func(resp http.ResponseWriter, req *http.Request) {
-		// Set CORS headers
-		resp.Header().Set("Access-Control-Allow-Origin", "*")
-		resp.Header().Set("Access-Control-Allow-Headers", fmt.Sprintf("Content-Type, x-grpc-web, %s", common.LAVA_CONSUMER_PROCESS_GUID))
-
-		if req.URL.Path == healthCheckPath && req.Method == http.MethodGet {
-			resp.WriteHeader(http.StatusOK)
-			resp.Write([]byte("Healthy"))
-			return
-		}
 
-		wrappedServer.ServeHTTP(resp, req)
+	// Wrap with TLS if enabled
+	if !networkAddress.DisableTLS {
+		tlsConfig := lavasession.GetTlsConfig(networkAddress)
+		lis = tls.NewListener(lis, tlsConfig)
+	} else {
+		utils.LavaFormatInfo("Running with disabled TLS configuration")
 	}
 
-	pl.httpServer = http.Server{
-		Handler: h2c.NewHandler(http.HandlerFunc(handler), &http2.Server{}),
-	}
+	// Create connection multiplexer to handle both HTTP and gRPC on same port
+	mux := cmux.New(lis)
+	// Match HTTP/1.1 first for health checks (fast prefix match)
+	httpListener := mux.Match(cmux.HTTP1Fast())
+	// Everything else goes to gRPC (avoids expensive header parsing)
+	grpcListener := mux.Match(cmux.Any())
+	pl.cmux = mux
 
-	var serveExecutor func() error
-	if networkAddress.DisableTLS {
-		utils.LavaFormatInfo("Running with disabled TLS configuration")
-		serveExecutor = func() error {
-			return pl.httpServer.Serve(lis)
-		}
-	} else {
-		pl.httpServer.TLSConfig = lavasession.GetTlsConfig(networkAddress)
-		serveExecutor = func() error {
-			return pl.httpServer.ServeTLS(lis, "", "")
-		}
+	// Build gRPC server
+	opts := []grpc.ServerOption{
+		grpc.MaxRecvMsgSize(1024 * 1024 * 512), // 512MB for large debug responses
+		grpc.MaxSendMsgSize(1024 * 1024 * 512), // 512MB for large debug responses
 	}
+	grpcServer := grpc.NewServer(opts...)
+	pl.grpcServer = grpcServer
 
+	// Register gRPC health checking service
+	healthServer := health.NewServer()
+	healthgrpc.RegisterHealthServer(grpcServer, healthServer)
+	healthServer.SetServingStatus("", healthgrpc.HealthCheckResponse_SERVING)
+	pl.healthServer = healthServer
+
+	// Register relay server
 	relayServer := &relayServer{relayReceivers: map[string]*relayReceiverWrapper{}}
 	pl.relayServer = relayServer
 	pairingtypes.RegisterRelayerServer(grpcServer, relayServer)
+
+	// Create HTTP server for health checks
+	httpMux := http.NewServeMux()
+	httpMux.HandleFunc(healthCheckPath, func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodGet {
+			w.WriteHeader(http.StatusOK)
+			w.Write([]byte("Healthy"))
+		} else {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+		}
+	})
+	httpServer := &http.Server{Handler: httpMux}
+	pl.httpServer = httpServer
+
+	// Start servers
 	go func() {
 		utils.LavaFormatInfo("New provider listener active", utils.Attribute{Key: "address", Value: networkAddress})
-		if err := serveExecutor(); !errors.Is(err, http.ErrServerClosed) {
-			utils.LavaFormatFatal("provider failed to serve", err, utils.Attribute{Key: "Address", Value: lis.Addr().String()})
+		if err := grpcServer.Serve(grpcListener); err != nil {
+			// Ignore expected shutdown errors
+			if isShutdownError(err) {
+				return
+			}
+			utils.LavaFormatFatal("gRPC server failed", err, utils.Attribute{Key: "Address", Value: networkAddress.Address})
+		}
+	}()
+
+	go func() {
+		if err := httpServer.Serve(httpListener); err != nil {
+			// Ignore expected shutdown errors
+			if isShutdownError(err) {
+				return
+			}
+			utils.LavaFormatFatal("HTTP health server failed", err, utils.Attribute{Key: "Address", Value: networkAddress.Address})
+		}
+	}()
+
+	go func() {
+		if err := mux.Serve(); err != nil {
+			if !isShutdownError(err) {
+				utils.LavaFormatError("cmux serve error", err)
+			}
 		}
 		utils.LavaFormatInfo("listener closed server", utils.Attribute{Key: "address", Value: networkAddress})
 	}()
+
 	return pl
 }