test: cover the ADR-0012 accelerate gate on the block path

pkg/config/config.go

@@ -135,15 +135,7 @@ func LoadConfig(confFile string) (*AppConfig, error) {
 	}
 	cfg.Version = version.GetVersion()
 	cfg.CommitHash = version.GetCommitHash()
-	cfg.propagationEnabled.Store(cfg.PropagationEnabledRaw)
-	cfg.skipMessageFromSelf.Store(true)
-	var aggMs int64
-	if cfg.AggregationIntervalMs == 0 {
-		aggMs = DefaultAggregationIntervalMs
-	} else {
-		aggMs = cfg.AggregationIntervalMs
-	}
-	cfg.aggregationIntervalMs.Store(aggMs)
+	cfg.InitDerived()
 
 	if err := cfg.Validate(); err != nil {
 		return nil, fmt.Errorf("failed to validate config: %w", err)
@@ -226,6 +218,23 @@ func (c *AppConfig) GetDCRotator() *commonconfig.Rotator {
 	return c.rotator
 }
 
+// InitDerived seeds atomics from yaml/env fields. Required for hand-built configs;
+// unsafe after dynamic-config rotation, which owns those atomics afterwards.
+func (c *AppConfig) InitDerived() {
+	c.propagationEnabled.Store(c.PropagationEnabledRaw)
+	c.skipMessageFromSelf.Store(true)
+	c.aggregationIntervalMs.Store(c.effectiveAggregationIntervalMs())
+}
+
+// effectiveAggregationIntervalMs resolves the zero-means-default rule shared by
+// InitDerived and Validate.
+func (c *AppConfig) effectiveAggregationIntervalMs() int64 {
+	if c.AggregationIntervalMs == 0 {
+		return DefaultAggregationIntervalMs
+	}
+	return c.AggregationIntervalMs
+}
+
 // Validate ensures the AppConfig has valid and complete values
 func (c *AppConfig) Validate() error {
 	if c.IdentityLibP2PDir == "" {
@@ -291,13 +300,7 @@ func (c *AppConfig) Validate() error {
 	if c.AggregationIntervalMs < 0 {
 		return fmt.Errorf("aggregation_interval_ms must be non-negative")
 	}
-	var effectiveAggMs int64
-	if c.AggregationIntervalMs == 0 {
-		effectiveAggMs = DefaultAggregationIntervalMs
-	} else {
-		effectiveAggMs = c.AggregationIntervalMs
-	}
-	if effectiveAggMs > maxAggregationIntervalMs {
+	if c.effectiveAggregationIntervalMs() > maxAggregationIntervalMs {
 		return fmt.Errorf("aggregation_interval_ms must be <= %d", maxAggregationIntervalMs)
 	}
 

pkg/service/gossipsub-gateway/accelerate_gate_test.go

@@ -0,0 +1,99 @@
+package gossipsub_gateway
+
+import (
+	"context"
+	"encoding/binary"
+	"encoding/hex"
+	"testing"
+	"time"
+
+	"github.com/golang/snappy"
+	"github.com/libp2p/go-libp2p"
+	libp2ppubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/stretchr/testify/require"
+
+	commonentities "github.com/getoptimum/optimum-common/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/test_utils"
+	"github.com/getoptimum/optimum-gateway/pkg/utils"
+)
+
+// blockAtSlot rewrites the gossip block's slot; DecodeBeaconBlockHeader must read it back.
+func blockAtSlot(t *testing.T, hexBlock string, slot uint64) []byte {
+	t.Helper()
+	raw, err := hex.DecodeString(hexBlock)
+	require.NoError(t, err)
+	ssz, err := utils.DecodeSnappy(raw, utils.MaxGossipPayloadSize)
+	require.NoError(t, err)
+	off := 4 + 96 // SSZ prefix + BLS signature
+	require.GreaterOrEqual(t, len(ssz), off+8, "fixture is too short to hold a slot")
+	binary.LittleEndian.PutUint64(ssz[off:off+8], slot)
+	encoded := snappy.Encode(nil, ssz)
+	hdr, err := consensus.DecodeBeaconBlockHeader(encoded)
+	require.NoError(t, err)
+	require.Equal(t, slot, hdr.Header.Slot, "slot rewrite landed at the wrong offset")
+	return encoded
+}
+
+// joinCLTopic subscribes to a real gossipsub topic so CL publishes can be read back.
+func joinCLTopic(t *testing.T, svc *Service, topic string) *libp2ppubsub.Subscription {
+	t.Helper()
+	h, err := libp2p.New(libp2p.NoListenAddrs)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = h.Close() })
+	ps, err := libp2ppubsub.NewGossipSub(t.Context(), h)
+	require.NoError(t, err)
+	tp, err := ps.Join(topic)
+	require.NoError(t, err)
+	sub, err := tp.Subscribe()
+	require.NoError(t, err)
+	t.Cleanup(sub.Cancel)
+	svc.libP2PTopics.Store(topic, tp)
+	return sub
+}
+
+// Unselected slot is withheld from the CL; both blocks are still streamed (gate is after arrival).
+func TestMumP2PBeaconBlockAccelerateGate(t *testing.T) {
+	svc, bootstrap := newGateway(t)
+	topic := "/eth2/deadbeef/beacon_block/ssz_snappy"
+	clSub := joinCLTopic(t, svc, topic)
+	hub := streamhub.New()
+	svc.streamHub = hub
+	sub := hub.Subscribe(4)
+	t.Cleanup(sub.Close)
+	t.Cleanup(svc.messagesMap.Close)
+
+	cur := chainstate.CurrentSlot(time.Now())
+	bootstrap.SetAccelerateResponse(map[string]any{
+		"to_slot":         cur + 10,
+		"slots":           []int64{int64(cur)},
+		"generated_at_ms": 1,
+	})
+	svc.srvMsgRouter.RefreshAccelerateSlots(t.Context())
+
+	fixture := test_utils.HoodiBeaconBlockMessage1
+	svc.processMumP2PMessage(svc.log, &commonentities.P2PMessage{
+		SourceNodeID: "peer-1", Topic: topic, Message: blockAtSlot(t, fixture, cur+1),
+	})
+	svc.processMumP2PMessage(svc.log, &commonentities.P2PMessage{
+		SourceNodeID: "peer-1", Topic: topic, Message: blockAtSlot(t, fixture, cur),
+	})
+
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	got, err := clSub.Next(ctx)
+	require.NoError(t, err, "on-list slot must reach the CL")
+	delivered, err := consensus.DecodeBeaconBlockHeader(got.Data)
+	require.NoError(t, err)
+	require.Equal(t, cur, delivered.Header.Slot, "examined but unselected slot must be withheld from the CL")
+
+	// Covers gossipsub delivering the two publishes in either order.
+	idle, cancelIdle := context.WithTimeout(t.Context(), 250*time.Millisecond)
+	defer cancelIdle()
+	_, err = clSub.Next(idle)
+	require.ErrorIs(t, err, context.DeadlineExceeded, "only the on-list slot may reach the CL")
+
+	require.Len(t, sub.Events(), 2, "both blocks are streamed regardless of the verdict")
+}

pkg/service/message_router/accelerate.go

@@ -40,7 +40,8 @@ func decideAccelerate(w *accelerateWindow, slot uint64) string {
 	return "not_on_list"
 }
 
-func (s *Service) pollAccelerateSlots(ctx context.Context) {
+// RefreshAccelerateSlots runs one poll and swaps the whole window. A failed poll keeps the previous one.
+func (s *Service) RefreshAccelerateSlots(ctx context.Context) {
 	chainID := s.authMgr.Chain()
 	if chainID == "" || s.cfg.RemoteBootstrapURL == "" {
 		return

pkg/service/message_router/accelerate_test.go

@@ -10,13 +10,9 @@ import (
 	"github.com/stretchr/testify/require"
 
 	commonentities "github.com/getoptimum/optimum-common/pkg/entities"
-	"github.com/getoptimum/optimum-gateway/pkg/service/message_router"
 )
 
 func TestShouldAccelerateBlock(t *testing.T) {
-	srv := newTestService(t, commonentities.GatewayTypePartner)
-	require.True(t, srv.ShouldAccelerateBlock(1), "no list fail-opens")
-
 	var fail atomic.Bool
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		require.Equal(t, "/api/v2/hoodi/accelerate_slots", r.URL.Path)
@@ -33,14 +29,17 @@ func TestShouldAccelerateBlock(t *testing.T) {
 	}))
 	t.Cleanup(ts.Close)
 
-	message_router.PollAccelerateSlotsForTest(t, srv, ts.URL)
+	srv := newTestServiceAt(t, commonentities.GatewayTypePartner, ts.URL)
+	require.True(t, srv.ShouldAccelerateBlock(1), "no list fail-opens")
+
+	srv.RefreshAccelerateSlots(t.Context())
 	require.True(t, srv.ShouldAccelerateBlock(100))
 	require.True(t, srv.ShouldAccelerateBlock(101))
 	require.False(t, srv.ShouldAccelerateBlock(110), "examined, not selected")
 	require.True(t, srv.ShouldAccelerateBlock(121), "past to_slot fail-opens")
 
 	fail.Store(true)
-	message_router.PollAccelerateSlotsForTest(t, srv, ts.URL)
+	srv.RefreshAccelerateSlots(t.Context())
 	require.False(t, srv.ShouldAccelerateBlock(110), "failed poll must not clear the list")
 	require.True(t, srv.ShouldAccelerateBlock(100))
 }

pkg/service/message_router/bg_sync.go

@@ -24,7 +24,7 @@ func (s *Service) bgSync(ctx context.Context) {
 			return
 		case <-ticker.C:
 			s.SetKnownValidators(s.authMgr.ValidatorIndexes())
-			s.pollAccelerateSlots(ctx)
+			s.RefreshAccelerateSlots(ctx)
 		}
 	}
 }

pkg/service/message_router/service_test.go

@@ -219,6 +219,11 @@ func TestService_ResolveValidatorChunkUsesSortedValidatorSet(t *testing.T) {
 
 func newTestService(t *testing.T, pairedWith commonentities.GatewayType, validators ...uint64) *message_router.Service {
 	t.Helper()
+	return newTestServiceAt(t, pairedWith, "dev-bootstrap.getoptimum.io", validators...)
+}
+
+func newTestServiceAt(t *testing.T, pairedWith commonentities.GatewayType, bootstrapURL string, validators ...uint64) *message_router.Service {
+	t.Helper()
 
 	cnt := test_utils.GetClean(t)
 	rig := test_utils.NewAuthTestRig(t, test_utils.WithClaimModifier(func(claims *jwks_verifier.Claims) {
@@ -231,7 +236,7 @@ func newTestService(t *testing.T, pairedWith commonentities.GatewayType, validat
 	require.NoError(t, err)
 
 	srv, err := message_router.NewService(t.Context(), &config.AppConfig{
-		RemoteBootstrapURL: "dev-bootstrap.getoptimum.io",
+		RemoteBootstrapURL: bootstrapURL,
 	}, cnt.Log, m)
 	require.NoError(t, err)
 	srv.SetKnownValidators(validators)

pkg/test_utils/jwt_auth_claims.go

@@ -66,7 +66,9 @@ func (r *AuthTestRig) AppCfg(t *testing.T) *config.AppConfig {
 		TelemetryPort:          48123,
 		GatewayClusterID:       "test-cluster",
 		TelemetryEnable:        true,
+		PropagationEnabledRaw:  true, // match the yaml-loaded test configs
 	}
+	cfg.InitDerived()
 	require.NoError(t, cfg.Validate())
 	return cfg
 }

pkg/test_utils/local_bootstrap_server.go

@@ -35,13 +35,14 @@ type ExposeNodesRequest struct {
 
 // LocalBootstrapServer is an in-process httptest bootstrap stub for gateway tests.
 type LocalBootstrapServer struct {
-	rig           *AuthTestRig
-	messages      *syncx.RWMap[string, any]
-	forksResponse *syncx.RWMap[string, any]
-	registerReqs  chan RegisterGatewayRequest
-	exposeReqs    chan ExposeNodesRequest
-	latencyReqs   chan BlockLatencyRequest
-	srv           *httptest.Server
+	rig                *AuthTestRig
+	messages           *syncx.RWMap[string, any]
+	forksResponse      *syncx.RWMap[string, any]
+	accelerateResponse *syncx.RWMap[string, any]
+	registerReqs       chan RegisterGatewayRequest
+	exposeReqs         chan ExposeNodesRequest
+	latencyReqs        chan BlockLatencyRequest
+	srv                *httptest.Server
 }
 
 func NewLocalBootstrapServerWithRig(t *testing.T, rig *AuthTestRig) *LocalBootstrapServer {
@@ -54,6 +55,7 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ
 
 	messages := syncx.NewRWMap[string, any]()
 	forksResponse := syncx.NewRWMap[string, any]()
+	accelerateResponse := syncx.NewRWMap[string, any]()
 	registerReqs := make(chan RegisterGatewayRequest, 32)
 	exposeReqs := make(chan ExposeNodesRequest, 32)
 	latencyReqs := make(chan BlockLatencyRequest, 32)
@@ -87,6 +89,12 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ
 		}
 		return c.JSON(forksResponse.LoadAll())
 	})
+	app.Get("/api/v2/:chain/accelerate_slots", func(c fiber.Ctx) error {
+		if rig != nil {
+			require.True(t, c.HasHeader("Authorization"))
+		}
+		return c.JSON(accelerateResponse.LoadAll())
+	})
 	app.Post(utils.BootstrapHandleBlockLatencyV2, func(c fiber.Ctx) error {
 		var payload entities.LatencyComparator
 		require.NoError(t, json.Unmarshal(c.Body(), &payload))
@@ -106,13 +114,14 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ
 	t.Cleanup(srv.Close)
 
 	return &LocalBootstrapServer{
-		rig:           rig,
-		srv:           srv,
-		forksResponse: forksResponse,
-		messages:      messages,
-		registerReqs:  registerReqs,
-		exposeReqs:    exposeReqs,
-		latencyReqs:   latencyReqs,
+		rig:                rig,
+		srv:                srv,
+		forksResponse:      forksResponse,
+		accelerateResponse: accelerateResponse,
+		messages:           messages,
+		registerReqs:       registerReqs,
+		exposeReqs:         exposeReqs,
+		latencyReqs:        latencyReqs,
 	}
 }
 
@@ -128,6 +137,10 @@ func (m *LocalBootstrapServer) SetForkResponse(payload map[string]any) {
 	m.forksResponse.Replace(payload)
 }
 
+func (m *LocalBootstrapServer) SetAccelerateResponse(payload map[string]any) {
+	m.accelerateResponse.Replace(payload)
+}
+
 func (m *LocalBootstrapServer) SetMessagesResponse(payload map[string]any) {
 	m.messages.Replace(payload)
 }