fix(message-router): restore poll status codes and prime the window at startup

pkg/service/gossipsub-gateway/accelerate_gate_test.go

@@ -56,7 +56,15 @@ func joinCLTopic(t *testing.T, svc *Service, topic string) *libp2ppubsub.Subscri
 
 // 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)
+	cur := chainstate.CurrentSlot(time.Now())
+	// Seed before the router exists: bgSync primes at startup and may land after the refresh.
+	svc, _ := newGateway(t, func(b *test_utils.LocalBootstrapServer) {
+		b.SetAccelerateResponse(map[string]any{
+			"to_slot":         cur + 10,
+			"slots":           []int64{int64(cur)},
+			"generated_at_ms": 1,
+		})
+	})
 	topic := "/eth2/deadbeef/beacon_block/ssz_snappy"
 	clSub := joinCLTopic(t, svc, topic)
 	hub := streamhub.New()
@@ -65,12 +73,6 @@ func TestMumP2PBeaconBlockAccelerateGate(t *testing.T) {
 	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

pkg/service/gossipsub-gateway/setup_test.go

@@ -10,7 +10,8 @@ import (
 	"github.com/getoptimum/optimum-gateway/pkg/test_utils"
 )
 
-func newGateway(t *testing.T) (*Service, *test_utils.LocalBootstrapServer) {
+// prepare seeds the stub before any service polls it.
+func newGateway(t *testing.T, prepare ...func(*test_utils.LocalBootstrapServer)) (*Service, *test_utils.LocalBootstrapServer) {
 	t.Helper()
 
 	cnt := test_utils.GetClean(t)
@@ -21,6 +22,9 @@ func newGateway(t *testing.T) (*Service, *test_utils.LocalBootstrapServer) {
 		"fork_digest": "deadbeef",
 		"future_fork": "DDEEFF00",
 	})
+	for _, p := range prepare {
+		p(bootstrap)
+	}
 	cfg := rig.AppCfg(t)
 	cfg.RemoteBootstrapURL = bootstrap.URL()
 	srvAuth, err := auth_token.New(t.Context(), cnt.Log, cfg)

pkg/service/message_router/accelerate.go

@@ -2,10 +2,10 @@ package message_router
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"time"
 
+	"github.com/getoptimum/optimum-common/pkg/logger"
 	commonnet "github.com/getoptimum/optimum-common/pkg/net"
 	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
 	"github.com/getoptimum/optimum-gateway/pkg/utils"
@@ -22,22 +22,29 @@ type accelerateSlotsResponse struct {
 	GeneratedAtMs int64   `json:"generated_at_ms"`
 }
 
+// Verdicts double as the `result` label on accelerate_decision_total.
+const (
+	accelerateOnList    = "on_list"
+	accelerateNotOnList = "not_on_list"
+	accelerateFailOpen  = "fail_open"
+)
+
 // ShouldAccelerateBlock is ADR-0012: accelerate unless the slot was examined and
 // not selected. Header slot, not the clock. No list / past to_slot fail-opens.
 func (s *Service) ShouldAccelerateBlock(slot uint64) bool {
 	decision := decideAccelerate(s.accelerate.Load(), slot)
 	telemetry.IncAccelerateDecision(decision)
-	return decision != "not_on_list"
+	return decision != accelerateNotOnList
 }
 
 func decideAccelerate(w *accelerateWindow, slot uint64) string {
 	if w == nil || w.toSlot == 0 || slot > w.toSlot {
-		return "fail_open"
+		return accelerateFailOpen
 	}
 	if _, ok := w.slots[slot]; ok {
-		return "on_list"
+		return accelerateOnList
 	}
-	return "not_on_list"
+	return accelerateNotOnList
 }
 
 // RefreshAccelerateSlots runs one poll and swaps the whole window. A failed poll keeps the previous one.
@@ -46,19 +53,23 @@ func (s *Service) RefreshAccelerateSlots(ctx context.Context) {
 	if chainID == "" || s.cfg.RemoteBootstrapURL == "" {
 		return
 	}
-	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
+	// Mint gets its own deadline: it runs on http.DefaultClient, so a hung auth stalls bgSync.
 	var headers map[string]string
-	if tok, err := s.authMgr.ServicesToken(ctx); err == nil && tok != "" {
+	tokCtx, cancelTok := context.WithTimeout(ctx, 5*time.Second)
+	tok, tokErr := s.authMgr.ServicesToken(tokCtx)
+	cancelTok()
+	if tokErr != nil {
+		s.log.Error("accelerate_slots poll has no services token, polling unauthenticated", tokErr)
+	}
+	if tok != "" {
 		headers = map[string]string{"Authorization": "Bearer " + tok}
 	}
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
 	res, code, err := commonnet.GetCurl[accelerateSlotsResponse](ctx, utils.BootstrapAccelerateSlotsURL(s.cfg.RemoteBootstrapURL, chainID), headers)
-	if err != nil {
-		s.log.Error("accelerate_slots poll failed, keeping previous list", err)
-		return
-	}
-	if code != http.StatusOK || res == nil {
-		s.log.Error("accelerate_slots poll failed, keeping previous list", fmt.Errorf("status code: %d", code))
+	// GetCurl reports a non-JSON body as an unmarshal error alongside the code.
+	if err != nil || code != http.StatusOK || res == nil {
+		s.log.Error("accelerate_slots poll failed, keeping previous list", err, logger.WithInt("status_code", code))
 		return
 	}
 	w := &accelerateWindow{slots: make(map[uint64]struct{}, len(res.Slots))}

pkg/service/message_router/accelerate_test.go

@@ -6,17 +6,21 @@ import (
 	"net/http/httptest"
 	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	commonentities "github.com/getoptimum/optimum-common/pkg/entities"
 )
 
 func TestShouldAccelerateBlock(t *testing.T) {
 	var fail atomic.Bool
+	fail.Store(true) // keep window nil until the fail-open assert; bgSync polls at start
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		require.Equal(t, "/api/v2/hoodi/accelerate_slots", r.URL.Path)
-		require.NotEmpty(t, r.Header.Get("Authorization"))
+		// assert, not require: FailNow off the test goroutine is unsupported.
+		assert.Equal(t, "/api/v2/hoodi/accelerate_slots", r.URL.Path)
+		assert.NotEmpty(t, r.Header.Get("Authorization"))
 		if fail.Load() {
 			w.WriteHeader(http.StatusInternalServerError)
 			return
@@ -32,6 +36,7 @@ func TestShouldAccelerateBlock(t *testing.T) {
 	srv := newTestServiceAt(t, commonentities.GatewayTypePartner, ts.URL)
 	require.True(t, srv.ShouldAccelerateBlock(1), "no list fail-opens")
 
+	fail.Store(false)
 	srv.RefreshAccelerateSlots(t.Context())
 	require.True(t, srv.ShouldAccelerateBlock(100))
 	require.True(t, srv.ShouldAccelerateBlock(101))
@@ -43,3 +48,21 @@ func TestShouldAccelerateBlock(t *testing.T) {
 	require.False(t, srv.ShouldAccelerateBlock(110), "failed poll must not clear the list")
 	require.True(t, srv.ShouldAccelerateBlock(100))
 }
+
+func TestAccelerateSlotsPrimedAtStartup(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"to_slot":         120,
+			"slots":           []int64{100},
+			"generated_at_ms": 1,
+		})
+	}))
+	t.Cleanup(ts.Close)
+
+	srv := newTestServiceAt(t, commonentities.GatewayTypePartner, ts.URL)
+
+	require.Eventually(t, func() bool {
+		return !srv.ShouldAccelerateBlock(110)
+	}, 5*time.Second, 5*time.Millisecond, "startup must fetch the window without waiting for a tick")
+	require.True(t, srv.ShouldAccelerateBlock(100), "selected slot still accelerates")
+}

pkg/service/message_router/bg_sync.go

@@ -16,6 +16,9 @@ import (
 // and the per-tick work degenerates to a single comparison, so we don't
 // bother short-circuiting here.
 func (s *Service) bgSync(ctx context.Context) {
+	// Prime so a restart is not fail-open until the first 30s tick.
+	s.RefreshAccelerateSlots(ctx)
+
 	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()
 	for {

pkg/service/message_router/service_test.go

@@ -3,6 +3,8 @@ package message_router_test
 import (
 	"bytes"
 	"encoding/hex"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 	"time"
 
@@ -219,7 +221,12 @@ 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...)
+	// bgSync polls at startup; an unstubbed URL would put every caller on the network.
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	t.Cleanup(ts.Close)
+	return newTestServiceAt(t, pairedWith, ts.URL, validators...)
 }
 
 func newTestServiceAt(t *testing.T, pairedWith commonentities.GatewayType, bootstrapURL string, validators ...uint64) *message_router.Service {

pkg/test_utils/local_bootstrap_server.go

@@ -84,14 +84,14 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ
 		return nil
 	})
 	app.Get(utils.BootstrapForkDigestPath, func(c fiber.Ctx) error {
-		if rig != nil {
-			require.True(t, c.HasHeader("Authorization"))
+		if err := requireAuth(rig, c); err != nil {
+			return err
 		}
 		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"))
+		if err := requireAuth(rig, c); err != nil {
+			return err
 		}
 		return c.JSON(accelerateResponse.LoadAll())
 	})
@@ -125,6 +125,14 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ
 	}
 }
 
+// Fiber handlers cannot require.FailNow; return 401 instead.
+func requireAuth(rig *AuthTestRig, c fiber.Ctx) error {
+	if rig == nil || c.HasHeader("Authorization") {
+		return nil
+	}
+	return c.Status(fiber.StatusUnauthorized).JSON(fiber.Map{"error": "missing Authorization header"})
+}
+
 func mapToURLValues(src map[string]string) url.Values {
 	dst := make(url.Values, len(src))
 	for key, value := range src {