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,8 +163,10 @@ 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()
@@ -173,6 +177,12 @@ func main() {
 			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
 			BufferSize:     appConf.StreamBufferSize,
 		}, l)
+		streamGRPCServer = stream.NewGRPCServer(hub, authenticator, stream.Config{
+			Addr:           appConf.StreamGRPCAddr,
+			MaxConns:       appConf.StreamMaxConns,
+			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
+			BufferSize:     appConf.StreamBufferSize,
+		}, l)
 	}
 
 	srvGateway, err := gateway.NewService(ctx, l, appConf, srvMessageRouter, authMgr, gateway.WithStreamHub(hub))
@@ -199,6 +209,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 +225,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/service/stream/grpc.go

@@ -0,0 +1,147 @@
+package stream
+
+import (
+	"context"
+	"net"
+	"strings"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"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
+}
+
+// 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: newConnLimiter(cfg.MaxConns, cfg.MaxConnsPerSub),
+		grpcSrv: grpc.NewServer(),
+	}
+	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
+				if err := stream.Send(&streamv1.BlockEvent{Lagged: true, Dropped: d}); 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 {
+	pe := &streamv1.BlockEvent{
+		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 {
+		pe.Raw = ev.Raw
+	}
+	return pe
+}
+
+// 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)
+}

pkg/service/stream/grpc_test.go

@@ -0,0 +1,139 @@
+package stream
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	"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/test_utils"
+)
+
+// newGRPCTestServer starts a GRPCServer over an in-memory bufconn and returns a
+// connected client. Auth is always required; loopback no-auth is covered by WS.
+func newGRPCTestServer(t *testing.T, cfg Config) (client streamv1.BlockStreamServiceClient, hub *streamhub.Service, rig *test_utils.AuthTestRig) {
+	t.Helper()
+	var authenticator ConsumerAuthenticator
+	authenticator, rig = testAuth(t, true)
+	hub = streamhub.New()
+	g := NewGRPCServer(hub, authenticator, cfg, logger.NewAppSLogger(logger.Debug))
+
+	lis := bufconn.Listen(1 << 20)
+	go func() { _ = g.grpcSrv.Serve(lis) }()
+	t.Cleanup(g.grpcSrv.Stop)
+
+	conn, err := grpc.NewClient("passthrough:///bufnet",
+		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+	return streamv1.NewBlockStreamServiceClient(conn), hub, rig
+}
+
+func authCtx(t *testing.T, rig *test_utils.AuthTestRig, subject string) context.Context {
+	t.Helper()
+	return metadata.AppendToOutgoingContext(context.Background(), "authorization", "Bearer "+streamToken(t, rig, subject))
+}
+
+func TestGRPC_RejectsWithoutToken(t *testing.T) {
+	client, hub, _ := newGRPCTestServer(t, Config{})
+
+	sub, err := client.Subscribe(context.Background(), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	_, err = sub.Recv()
+	require.Equal(t, codes.Unauthenticated, status.Code(err))
+	require.Zero(t, hub.SubscriberCount(), "rejected consumer must not create a subscriber")
+}
+
+func TestGRPC_DeliversFraming(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		mode      string
+		expectRaw bool
+	}{
+		{"metadata omits raw", "metadata", false},
+		{"raw includes bytes", "raw", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client, hub, rig := newGRPCTestServer(t, Config{})
+			sub, err := client.Subscribe(authCtx(t, rig, "sub-1"), &streamv1.SubscribeRequest{Mode: tc.mode})
+			require.NoError(t, err)
+
+			waitSubscribed(t, hub, 1)
+			hub.Emit(sampleEvent())
+
+			ev, err := sub.Recv()
+			require.NoError(t, err)
+			require.EqualValues(t, 42, ev.GetSlot())
+			require.False(t, ev.GetLagged())
+			if tc.expectRaw {
+				require.Equal(t, []byte("ssz-snappy-bytes"), ev.GetRaw())
+			} else {
+				require.Empty(t, ev.GetRaw())
+			}
+		})
+	}
+}
+
+func TestGRPC_LaggedOnOverflow(t *testing.T) {
+	client, hub, rig := newGRPCTestServer(t, Config{BufferSize: 1})
+	sub, err := client.Subscribe(authCtx(t, rig, "sub-1"), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+
+	waitSubscribed(t, hub, 1)
+	for range 3000 {
+		hub.Emit(sampleEvent())
+	}
+
+	var sawLagged bool
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) && !sawLagged {
+		ev, rerr := sub.Recv()
+		require.NoError(t, rerr)
+		if ev.GetLagged() {
+			require.Positive(t, ev.GetDropped())
+			sawLagged = true
+		}
+	}
+	require.True(t, sawLagged, "a lagged frame must be sent after overflow")
+}
+
+func TestGRPC_GlobalCapRejects(t *testing.T) {
+	client, hub, rig := newGRPCTestServer(t, Config{MaxConns: 1})
+
+	first, err := client.Subscribe(authCtx(t, rig, "sub-a"), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	waitSubscribed(t, hub, 1)
+
+	second, err := client.Subscribe(authCtx(t, rig, "sub-b"), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	_, err = second.Recv()
+	require.Equal(t, codes.ResourceExhausted, status.Code(err))
+
+	_ = first // keep the first stream open for the duration of the assertion
+}
+
+func TestGRPC_CleanupOnCancel(t *testing.T) {
+	client, hub, rig := newGRPCTestServer(t, Config{})
+	ctx, cancel := context.WithCancel(authCtx(t, rig, "sub-1"))
+	_, err := client.Subscribe(ctx, &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+
+	waitSubscribed(t, hub, 1)
+	cancel()
+
+	// Cancel must unwind the handler, closing the subscriber (drop-counter entry
+	// included) and releasing the cap slot, so nothing leaks.
+	waitSubscribed(t, hub, 0)
+}