feat: add consumer block-stream gRPC transport

buf.gen.yaml

@@ -4,3 +4,7 @@ plugins:
     out: .
     opt:
       - module=github.com/getoptimum/optimum-gateway
+  - local: ["go", "tool", "protoc-gen-go-grpc"]
+    out: .
+    opt:
+      - module=github.com/getoptimum/optimum-gateway

cmd/main.go

@@ -12,6 +12,8 @@ import (
 	"syscall"
 	"time"
 
+	"google.golang.org/grpc"
+
 	commonio "github.com/getoptimum/optimum-common/pkg/io"
 	"github.com/getoptimum/optimum-common/pkg/logger"
 	"github.com/getoptimum/optimum-gateway/pkg/config"
@@ -161,17 +163,30 @@ func main() {
 	}
 
 	// Consumer block-stream (ADR-0011), opt-in and off by default. The hub must
-	// exist before the gateway so it can be wired as an emit sink.
+	// exist before the gateway so it can be wired as an emit sink; WS and gRPC
+	// share the one hub.
 	var streamServer *stream.Server
+	var streamGRPCServer *stream.GRPCServer
 	var hub *streamhub.Service
 	if appConf.StreamEnable {
 		hub = streamhub.New()
 		authenticator := stream.NewConsumerAuthenticator(authMgr, appConf.StreamRequireAuth)
+		// One limiter across both transports keeps the caps global, not
+		// per-transport; config validation guarantees the caps are > 0.
+		limiter := stream.NewConnLimiter(appConf.StreamMaxConns, appConf.StreamMaxConnsPerSub)
 		streamServer = stream.NewServer(hub, authenticator, stream.Config{
 			Addr:           appConf.StreamAddr,
 			MaxConns:       appConf.StreamMaxConns,
 			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
 			BufferSize:     appConf.StreamBufferSize,
+			Limiter:        limiter,
+		}, l)
+		streamGRPCServer = stream.NewGRPCServer(hub, authenticator, stream.Config{
+			Addr:           appConf.StreamGRPCAddr,
+			MaxConns:       appConf.StreamMaxConns,
+			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
+			BufferSize:     appConf.StreamBufferSize,
+			Limiter:        limiter,
 		}, l)
 	}
 
@@ -199,6 +214,14 @@ func main() {
 		}()
 	}
 
+	if streamGRPCServer != nil {
+		go func() {
+			if runErr := streamGRPCServer.Run(); runErr != nil && !errors.Is(runErr, grpc.ErrServerStopped) {
+				l.Fatal("failed to run consumer stream grpc server", runErr)
+			}
+		}()
+	}
+
 	<-c // This blocks the main thread until an interrupt is received
 	cancel()
 	_ = appRouter.Stop()
@@ -207,6 +230,9 @@ func main() {
 		_ = streamServer.Stop(shutdownCtx)
 		cancelShutdown()
 	}
+	if streamGRPCServer != nil {
+		streamGRPCServer.Stop()
+	}
 	srvGateway.Stop()
 	if lokiDone != nil {
 		<-lokiDone // wait for final Loki flush to complete

go.mod

@@ -5,6 +5,7 @@ go 1.26.5
 tool (
 	github.com/bufbuild/buf/cmd/buf
 	github.com/golangci/golangci-lint/v2/cmd/golangci-lint
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc
 	google.golang.org/protobuf/cmd/protoc-gen-go
 )
 
@@ -31,6 +32,7 @@ require (
 	github.com/prometheus/common v0.70.1
 	github.com/prometheus/prometheus v0.311.3
 	github.com/stretchr/testify v1.11.1
+	google.golang.org/grpc v1.82.1
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v2 v2.4.0
 )
@@ -453,7 +455,7 @@ require (
 	google.golang.org/api v0.272.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260523011958-0a33c5d7ca68 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68 // indirect
-	google.golang.org/grpc v1.82.1 // indirect
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	honnef.co/go/tools v0.7.0 // indirect

go.sum

@@ -1275,6 +1275,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68 h1:
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
 google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
 google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

pkg/config/config.go

@@ -263,6 +263,9 @@ func (c *AppConfig) Validate() error {
 		if err := validateStreamListener("stream_grpc_addr", c.StreamGRPCAddr, c.StreamRequireAuth); err != nil {
 			return err
 		}
+		if strings.TrimSpace(c.StreamAddr) == strings.TrimSpace(c.StreamGRPCAddr) {
+			return fmt.Errorf("stream_addr and stream_grpc_addr must differ, got %q", c.StreamAddr)
+		}
 		if c.StreamMaxConns <= 0 {
 			return fmt.Errorf("stream_max_conns must be > 0")
 		}

pkg/service/stream/grpc.go

@@ -0,0 +1,157 @@
+package stream
+
+import (
+	"context"
+	"net"
+	"strings"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"github.com/getoptimum/optimum-common/pkg/logger"
+	streamv1 "github.com/getoptimum/optimum-gateway/pkg/service/stream/v1"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
+	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
+)
+
+// GRPCServer serves the consumer block-stream over gRPC on its own listener,
+// reusing the hub, authenticator, and connection caps of the WS transport.
+type GRPCServer struct {
+	streamv1.UnimplementedBlockStreamServiceServer
+	hub     *streamhub.Service
+	auth    ConsumerAuthenticator
+	cfg     Config
+	log     logger.AppLogger
+	limiter *ConnLimiter
+	grpcSrv *grpc.Server
+}
+
+// maxConcurrentStreams bounds Subscribe streams per connection. The caps run
+// after auth, so one unauthenticated socket would otherwise be unbounded.
+const maxConcurrentStreams = 256
+
+// NewGRPCServer builds the consumer gRPC server. It does not start listening;
+// call Run.
+func NewGRPCServer(hub *streamhub.Service, auth ConsumerAuthenticator, cfg Config, log logger.AppLogger) *GRPCServer {
+	cfg = withDefaults(cfg)
+	g := &GRPCServer{
+		hub:     hub,
+		auth:    auth,
+		cfg:     cfg,
+		log:     log.With(logger.WithService("stream-grpc")),
+		limiter: cfg.Limiter,
+		grpcSrv: grpc.NewServer(
+			// Reap dead peers on the WS clock; the gRPC default is a 2h ping.
+			grpc.KeepaliveParams(keepalive.ServerParameters{Time: pingPeriod, Timeout: writeWait}),
+			grpc.MaxConcurrentStreams(maxConcurrentStreams),
+		),
+	}
+	streamv1.RegisterBlockStreamServiceServer(g.grpcSrv, g)
+	return g
+}
+
+// Run serves until Stop is called; grpc.ErrServerStopped on a clean stop is
+// treated as normal by the caller.
+func (g *GRPCServer) Run() error {
+	lis, err := net.Listen("tcp", g.cfg.Addr)
+	if err != nil {
+		return err
+	}
+	g.log.Info("starting consumer stream grpc server", logger.WithString("addr", g.cfg.Addr))
+	return g.grpcSrv.Serve(lis)
+}
+
+// Stop hard-stops the server, canceling active Subscribe streams so shutdown
+// does not block on long-lived consumers.
+func (g *GRPCServer) Stop() { g.grpcSrv.Stop() }
+
+// Subscribe authenticates and enforces caps before opening the stream, then
+// drains the buffer as proto frames (metadata omits Raw); lagged on overflow.
+func (g *GRPCServer) Subscribe(req *streamv1.SubscribeRequest, stream grpc.ServerStreamingServer[streamv1.BlockEvent]) error {
+	mode, ok := normalizeMode(req.GetMode())
+	if !ok {
+		return status.Error(codes.InvalidArgument, "invalid mode")
+	}
+	if !topicsOK(req.GetTopics()...) {
+		return status.Error(codes.InvalidArgument, "unsupported topics")
+	}
+
+	ctx := stream.Context()
+	subject, err := g.auth.Authenticate(metadataToken(ctx))
+	if err != nil {
+		telemetry.RecordStreamAuthFailure()
+		return status.Error(codes.Unauthenticated, "unauthorized")
+	}
+	if !g.limiter.acquire(subject) {
+		return status.Error(codes.ResourceExhausted, "too many connections")
+	}
+	defer g.limiter.release(subject)
+
+	sub := g.hub.Subscribe(g.cfg.BufferSize)
+	defer sub.Close()
+
+	raw := mode == modeRaw
+	var lastDropped uint64
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case ev, ok := <-sub.Events():
+			if !ok {
+				return nil
+			}
+			if d := sub.Dropped(); d != lastDropped {
+				lastDropped = d
+				lag := &streamv1.BlockEvent{Frame: &streamv1.BlockEvent_Lagged{Lagged: &streamv1.Lagged{Dropped: d}}}
+				if err := stream.Send(lag); err != nil {
+					return err
+				}
+			}
+			if err := stream.Send(toProto(ev, raw)); err != nil {
+				return err
+			}
+			telemetry.RecordStreamEventSent()
+		}
+	}
+}
+
+func toProto(ev *streamhub.BlockEvent, raw bool) *streamv1.BlockEvent {
+	b := &streamv1.Block{
+		Slot:           ev.Slot,
+		ProposerIndex:  ev.ProposerIndex,
+		ParentRoot:     ev.ParentRoot,
+		StateRoot:      ev.StateRoot,
+		BlockSizeBytes: ev.BlockSizeBytes,
+		Topic:          ev.Topic,
+		Source:         string(ev.Source),
+		ReceivedAtMs:   ev.ReceivedAtMs,
+		GatewayId:      ev.GatewayID,
+		ForkDigest:     ev.ForkDigest,
+		Stale:          ev.Stale,
+	}
+	if raw {
+		b.Raw = ev.Raw
+	}
+	return &streamv1.BlockEvent{Frame: &streamv1.BlockEvent_Block{Block: b}}
+}
+
+// metadataToken reads the consumer JWT from the "authorization" gRPC metadata,
+// accepting either a bare token or a "Bearer <jwt>" value.
+func metadataToken(ctx context.Context) string {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return ""
+	}
+	vals := md.Get("authorization")
+	if len(vals) == 0 {
+		return ""
+	}
+	tok := vals[0]
+	if after, ok := strings.CutPrefix(tok, "Bearer "); ok {
+		tok = after
+	}
+	return strings.TrimSpace(tok)
+}