feat: add consumer block-stream hub

pkg/config/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -90,6 +91,16 @@ type AppConfig struct {
 	// gateway logs a loud startup warning when off.
 	EnableAuth bool `yaml:"enable_auth" env:"OPT_ENABLE_AUTH" default:"true"`
 
+	// Consumer block-stream (ADR-0011): opt-in read-only fan-out of decoded
+	// beacon blocks over WebSocket/gRPC, off by default.
+	StreamEnable         bool   `yaml:"stream_enable" env:"OPT_STREAM_ENABLE" default:"false"`
+	StreamAddr           string `yaml:"stream_addr" env:"OPT_STREAM_ADDR" default:"0.0.0.0:9600"`
+	StreamGRPCAddr       string `yaml:"stream_grpc_addr" env:"OPT_STREAM_GRPC_ADDR" default:"0.0.0.0:9601"`
+	StreamRequireAuth    bool   `yaml:"stream_require_auth" env:"OPT_STREAM_REQUIRE_AUTH" default:"true"`
+	StreamMaxConns       int    `yaml:"stream_max_conns" env:"OPT_STREAM_MAX_CONNS" default:"256"`
+	StreamMaxConnsPerSub int    `yaml:"stream_max_conns_per_sub" env:"OPT_STREAM_MAX_CONNS_PER_SUB" default:"8"`
+	StreamBufferSize     int    `yaml:"stream_buffer_size" env:"OPT_STREAM_BUFFER_SIZE" default:"64"`
+
 	RemotePushEnable   bool   `yaml:"remote_push_enable" env:"OPT_REMOTE_PUSH_ENABLE" default:"false"`
 	RemotePushMimirURL string `yaml:"remote_push_mimir_url" env:"OPT_REMOTE_PUSH_MIMIR_URL" default:"https://v2-mimir.getoptimum.io"`
 	RemotePushLokiURL  string `yaml:"remote_push_loki_url" env:"OPT_REMOTE_PUSH_LOKI_URL" default:"https://v2-loki.getoptimum.io"`
@@ -245,6 +256,24 @@ func (c *AppConfig) Validate() error {
 		return fmt.Errorf("OPT_GATEWAY_CLUSTER_ID is required")
 	}
 
+	if c.StreamEnable {
+		if err := validateStreamListener("stream_addr", c.StreamAddr, c.StreamRequireAuth); err != nil {
+			return err
+		}
+		if err := validateStreamListener("stream_grpc_addr", c.StreamGRPCAddr, c.StreamRequireAuth); err != nil {
+			return err
+		}
+		if c.StreamMaxConns <= 0 {
+			return fmt.Errorf("stream_max_conns must be > 0")
+		}
+		if c.StreamMaxConnsPerSub <= 0 {
+			return fmt.Errorf("stream_max_conns_per_sub must be > 0")
+		}
+		if c.StreamBufferSize <= 0 {
+			return fmt.Errorf("stream_buffer_size must be > 0")
+		}
+	}
+
 	if c.AggregationIntervalMs < 0 {
 		return fmt.Errorf("aggregation_interval_ms must be non-negative")
 	}
@@ -261,6 +290,34 @@ func (c *AppConfig) Validate() error {
 	return nil
 }
 
+// validateStreamListener enforces the ADR-0011 exposure rule: auth may be
+// disabled only on a loopback bind.
+func validateStreamListener(field, addr string, requireAuth bool) error {
+	host, port, err := net.SplitHostPort(strings.TrimSpace(addr))
+	if err != nil {
+		return fmt.Errorf("invalid %s %q: %w", field, addr, err)
+	}
+	if p, perr := strconv.Atoi(port); perr != nil || p <= 0 || p > 65535 {
+		return fmt.Errorf("invalid %s %q: port must be between 1 and 65535", field, addr)
+	}
+	if !requireAuth && !isLoopbackHost(host) {
+		return fmt.Errorf("%s=%q requires stream_require_auth=true (auth may be disabled only on a loopback bind)", field, addr)
+	}
+	return nil
+}
+
+// isLoopbackHost treats an empty host (binds all interfaces) as non-loopback.
+func isLoopbackHost(host string) bool {
+	switch host {
+	case "":
+		return false
+	case "localhost":
+		return true
+	}
+	ip := net.ParseIP(host)
+	return ip != nil && ip.IsLoopback()
+}
+
 func (c *AppConfig) PropagationEnabled() bool {
 	return c.propagationEnabled.Load()
 }

pkg/config/config_test.go

@@ -269,3 +269,44 @@ gateway_id: local-dockerized
 	require.Equal(t, "local-dockerized", cfg.GatewayID)
 	require.Equal(t, "optimum_hoodi_v0_1", cfg.GatewayClusterID)
 }
+
+// The stream is off by default, and when enabled auth may be disabled only on
+// a loopback bind (ADR-0011 exposure rule).
+func TestStreamValidation(t *testing.T) {
+	base := func(t *testing.T) {
+		t.Helper()
+		t.Setenv("OPT_IDENTITY_LIBP2P_DIR", "./libid")
+		t.Setenv("OPT_IDENTITY_MUMP2P_DIR", "./mump2pid")
+		t.Setenv("OPT_AGENT_LIB_P2P_PORT", "5000")
+		t.Setenv("OPT_AGENT_MUMP2P_PORT", "5001")
+		t.Setenv("OPT_GATEWAY_CLUSTER_ID", "gw-cluster")
+		t.Setenv("OPT_TELEMETRY_PORT", "8888")
+	}
+
+	t.Run("off by default", func(t *testing.T) {
+		base(t)
+		cfg, err := config.LoadConfig("")
+		require.NoError(t, err)
+		require.False(t, cfg.StreamEnable)
+		require.True(t, cfg.StreamRequireAuth)
+	})
+
+	t.Run("auth off on loopback allowed", func(t *testing.T) {
+		base(t)
+		t.Setenv("OPT_STREAM_ENABLE", "true")
+		t.Setenv("OPT_STREAM_REQUIRE_AUTH", "false")
+		t.Setenv("OPT_STREAM_ADDR", "127.0.0.1:9600")
+		t.Setenv("OPT_STREAM_GRPC_ADDR", "localhost:9601")
+		_, err := config.LoadConfig("")
+		require.NoError(t, err)
+	})
+
+	t.Run("auth off on exposed bind rejected", func(t *testing.T) {
+		base(t)
+		t.Setenv("OPT_STREAM_ENABLE", "true")
+		t.Setenv("OPT_STREAM_REQUIRE_AUTH", "false")
+		t.Setenv("OPT_STREAM_ADDR", "0.0.0.0:9600")
+		_, err := config.LoadConfig("")
+		require.ErrorContains(t, err, "stream_require_auth=true")
+	})
+}

pkg/service/gossipsub-gateway/beacon_block_measures.go

@@ -7,6 +7,7 @@ import (
 	"github.com/getoptimum/optimum-gateway/pkg/entities"
 	chainstate "github.com/getoptimum/optimum-gateway/pkg/protocol/chain_state"
 	"github.com/getoptimum/optimum-gateway/pkg/protocol/consensus"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
 	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
 	"github.com/getoptimum/optimum-gateway/pkg/utils"
 )
@@ -47,7 +48,28 @@ func (s *Service) processBeaconBlockArrival(
 
 	// stale blocks are still measured above but not forwarded, to avoid polluting the mesh
 	currentSlot := chainstate.CurrentSlot(time.Now())
-	if diff := utils.DiffUint64(blockDecoded.Header.Slot, currentSlot); diff > staleSlotThreshold {
+	diff := utils.DiffUint64(blockDecoded.Header.Slot, currentSlot)
+	stale := diff > staleSlotThreshold
+
+	// Stream every observation, stale flagged rather than dropped (ADR-0011).
+	if s.streamHub != nil {
+		s.streamHub.Emit(&streamhub.BlockEvent{
+			Slot:           blockDecoded.Header.Slot,
+			ProposerIndex:  blockDecoded.Header.ProposerIndex,
+			ParentRoot:     blockDecoded.Header.ParentRoot,
+			StateRoot:      blockDecoded.Header.StateRoot,
+			BlockSizeBytes: uint64(len(msg)),
+			Topic:          topic,
+			Source:         source,
+			ReceivedAtMs:   recvAt,
+			GatewayID:      s.cfg.GatewayID,
+			ForkDigest:     s.srvForkMgr.ActiveDigest(),
+			Stale:          stale,
+			Raw:            msg,
+		})
+	}
+
+	if stale {
 		l.Info("stale_data, skipping publish of beacon_block",
 			logger.WithUint64("current_slot", currentSlot),
 			logger.WithUint64("diff", diff),

pkg/service/gossipsub-gateway/service.go

@@ -24,6 +24,7 @@ import (
 	"github.com/getoptimum/optimum-gateway/pkg/service/bootstrapper"
 	"github.com/getoptimum/optimum-gateway/pkg/service/message_router"
 	"github.com/getoptimum/optimum-gateway/pkg/service/mum_p2p"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
 	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry/tracer"
 )
 
@@ -76,6 +77,8 @@ type Service struct {
 
 	lastBlockReceivedAt atomic.Int64 // Unix ms — stamped on every beacon block from any source
 	startedAt           time.Time    // wall-clock time the service was created
+
+	streamHub *streamhub.Hub // consumer block-stream fan-out (ADR-0011); nil disables it
 }
 
 // LastBlockReceivedMs returns Unix ms of the last beacon block seen (0 if none).
@@ -94,6 +97,14 @@ func WithCustomMumP2PConnectionGater(gater connmgr.ConnectionGater) func(*Servic
 	}
 }
 
+// WithStreamHub wires the consumer block-stream fan-out (ADR-0011); nil (the
+// default) leaves the stream off.
+func WithStreamHub(hub *streamhub.Hub) Option {
+	return func(s *Service) {
+		s.streamHub = hub
+	}
+}
+
 func NewService(
 	ctx context.Context,
 	log logger.AppLogger,

pkg/service/gossipsub-gateway/stream_emit_test.go

@@ -0,0 +1,41 @@
+package gossipsub_gateway
+
+import (
+	"encoding/hex"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/getoptimum/optimum-gateway/pkg/entities"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
+	"github.com/getoptimum/optimum-gateway/pkg/test_utils"
+)
+
+// A decoded beacon block is emitted to the hub with its metadata and raw bytes.
+func TestProcessBeaconBlockArrivalEmitsToStreamHub(t *testing.T) {
+	svc, _ := newGateway(t)
+	hub := streamhub.New()
+	svc.streamHub = hub
+	sub := hub.Subscribe(4)
+
+	raw, err := hex.DecodeString(test_utils.HoodiBeaconBlockMessage1)
+	require.NoError(t, err)
+	topic := "/eth2/deadbeef/beacon_block/ssz_snappy"
+
+	slot, _ := svc.processBeaconBlockArrival(svc.log, topic, raw, time.Now().UnixMilli(), entities.SourceLibP2P, "", "peer-x")
+	require.Equal(t, uint64(3435697), slot)
+
+	select {
+	case ev := <-sub.Events():
+		require.Equal(t, uint64(3435697), ev.Slot)
+		require.Equal(t, uint64(526417), ev.ProposerIndex)
+		require.Equal(t, entities.SourceLibP2P, ev.Source)
+		require.Equal(t, topic, ev.Topic)
+		require.Equal(t, "deadbeef", ev.ForkDigest)
+		require.Equal(t, raw, ev.Raw)
+		require.True(t, ev.Stale, "an old-slot fixture is flagged stale but still streamed")
+	default:
+		t.Fatal("expected a block event on the hub")
+	}
+}

pkg/service/streamhub/event.go

@@ -0,0 +1,25 @@
+// Package streamhub fans decoded beacon-block observations out to downstream
+// consumers (ADR-0011). Emit is non-blocking: a slow consumer's oldest events
+// are dropped rather than backpressuring the ingest goroutines.
+package streamhub
+
+import "github.com/getoptimum/optimum-gateway/pkg/entities"
+
+// BlockEvent is one beacon-block observation. It is produced once per source,
+// so (Slot, ProposerIndex) is the block identity and Source tells the libp2p
+// and mump2p views apart. Raw holds the verbatim ssz_snappy bytes for raw-mode
+// consumers; metadata-mode transports omit it.
+type BlockEvent struct {
+	Slot           uint64
+	ProposerIndex  uint64
+	ParentRoot     []byte
+	StateRoot      []byte
+	BlockSizeBytes uint64
+	Topic          string
+	Source         entities.Source
+	ReceivedAtMs   int64
+	GatewayID      string
+	ForkDigest     string
+	Stale          bool
+	Raw            []byte
+}

pkg/service/streamhub/hub.go

@@ -0,0 +1,97 @@
+package streamhub
+
+import (
+	"sync"
+	"sync/atomic"
+
+	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
+)
+
+// DefaultBufferSize is the per-subscriber ring depth used when Subscribe is
+// given a non-positive size.
+const DefaultBufferSize = 64
+
+// Hub broadcasts each BlockEvent to all subscribers.
+type Hub struct {
+	mu   sync.RWMutex
+	subs map[*Subscription]struct{}
+}
+
+func New() *Hub {
+	return &Hub{subs: make(map[*Subscription]struct{})}
+}
+
+// Subscribe registers a consumer with a bounded ring of bufSize events. The
+// caller drains Events() and must Close() when done.
+func (h *Hub) Subscribe(bufSize int) *Subscription {
+	if bufSize <= 0 {
+		bufSize = DefaultBufferSize
+	}
+	sub := &Subscription{hub: h, events: make(chan *BlockEvent, bufSize)}
+	h.mu.Lock()
+	h.subs[sub] = struct{}{}
+	h.mu.Unlock()
+	return sub
+}
+
+// Emit broadcasts ev without blocking; the event is shared read-only, so
+// callers must not mutate it afterwards.
+func (h *Hub) Emit(ev *BlockEvent) {
+	h.mu.RLock()
+	for sub := range h.subs {
+		sub.offer(ev)
+	}
+	h.mu.RUnlock()
+}
+
+// Subscription is one consumer's bounded, drop-oldest view of the stream.
+type Subscription struct {
+	hub     *Hub
+	events  chan *BlockEvent
+	mu      sync.Mutex // serializes concurrent emits' evict+send
+	dropped atomic.Uint64
+}
+
+// Events is the read side of the ring; Close() closes it.
+func (s *Subscription) Events() <-chan *BlockEvent { return s.events }
+
+// Dropped is the cumulative count of events dropped on overflow.
+func (s *Subscription) Dropped() uint64 { return s.dropped.Load() }
+
+// Close unregisters the subscriber and closes its channel. Holding the hub
+// write lock guarantees no Emit is mid-send when the channel closes.
+func (s *Subscription) Close() {
+	s.hub.mu.Lock()
+	if _, ok := s.hub.subs[s]; ok {
+		delete(s.hub.subs, s)
+		close(s.events)
+	}
+	s.hub.mu.Unlock()
+}
+
+// offer enqueues ev, evicting the oldest event when the ring is full.
+func (s *Subscription) offer(ev *BlockEvent) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	select {
+	case s.events <- ev:
+		return
+	default:
+	}
+	select {
+	case <-s.events:
+		s.recordDrop()
+	default:
+	}
+	select {
+	case s.events <- ev:
+	default:
+		s.recordDrop()
+	}
+}
+
+func (s *Subscription) recordDrop() {
+	s.dropped.Add(1)
+	telemetry.RecordStreamEventDropped()
+}