From 134b3614b6a413b2074a247d5655e77b62fa029d Mon Sep 17 00:00:00 2001 From: Chris Fuka Date: Thu, 27 Aug 2026 10:36:07 -0500 Subject: [PATCH 1/2] fix: restore the poll's status code, and prime the window at startup Three things, each verified present before fixing. Status code. GetCurl reports a non-JSON body as an unmarshal error alongside the real code, so splitting the branches sent the "endpoint not deployed" case down the err path and dropped the code. A bootstrap without the route serves a 404 HTML page, and the log read only "failed to unmarshal response: invalid character '<'", which says nothing about why. status_code is now a field on the one branch, and the logger already tolerates a nil error, so the two cases fold back together. Startup prime. Nothing polled at construction, so a gateway accelerated every block until the first 30s tick: a slot inside the horizon and unselected was forwarded anyway. That is the fail-open path taken for want of an answer rather than because one was unavailable. bgSync now polls before entering its loop. TestShouldAccelerateBlock serves failures until its fail-open assertion is done, since the prime would otherwise race it. Token deadline. ServicesToken mints over HTTP when nothing is cached and was called with the poll's 5s context, so a slow mint ate the poll's budget. It now runs before the deadline is applied. Also, the stub's auth check answered with require on fiber's handler goroutine, where a failed assertion calls t.FailNow off the test goroutine. Go does not support that: the handler goexits mid-request and the caller sees a transport error rather than the missing header. Both routes now answer 401. Not addressed: the gateway still treats every slot below to_slot as examined, though bootstrap only examines [to_slot-96, to_slot]. A slot far beneath the window reads as "not selected" instead of "not looked at". Unreachable in normal operation, and not fixable here without from_slot in the response. --- pkg/service/message_router/accelerate.go | 19 +++++++------ pkg/service/message_router/accelerate_test.go | 28 +++++++++++++++++++ pkg/service/message_router/bg_sync.go | 5 ++++ pkg/test_utils/local_bootstrap_server.go | 19 ++++++++++--- 4 files changed, 58 insertions(+), 13 deletions(-) diff --git a/pkg/service/message_router/accelerate.go b/pkg/service/message_router/accelerate.go index 973a425..710f8aa 100644 --- a/pkg/service/message_router/accelerate.go +++ b/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" @@ -46,19 +46,20 @@ func (s *Service) RefreshAccelerateSlots(ctx context.Context) { if chainID == "" || s.cfg.RemoteBootstrapURL == "" { return } - ctx, cancel := context.WithTimeout(ctx, 5*time.Second) - defer cancel() + // Token first, outside the deadline below: ServicesToken mints over HTTP when + // nothing is cached, and that must not eat the poll's budget. var headers map[string]string if tok, err := s.authMgr.ServicesToken(ctx); err == nil && 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)) + // status_code is a field rather than part of err: a bootstrap without the + // endpoint serves a non-JSON 404, which GetCurl reports as an unmarshal error + // alongside the code, and the code is the half that says what is wrong. + 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))} diff --git a/pkg/service/message_router/accelerate_test.go b/pkg/service/message_router/accelerate_test.go index 0674aff..fea1626 100644 --- a/pkg/service/message_router/accelerate_test.go +++ b/pkg/service/message_router/accelerate_test.go @@ -6,6 +6,7 @@ import ( "net/http/httptest" "sync/atomic" "testing" + "time" "github.com/stretchr/testify/require" @@ -14,6 +15,9 @@ import ( func TestShouldAccelerateBlock(t *testing.T) { var fail atomic.Bool + // bgSync primes on startup, so serve failures until the fail-open assertion + // below is done. A failed poll leaves the window nil, which is what it needs. + fail.Store(true) 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")) @@ -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,26 @@ func TestShouldAccelerateBlock(t *testing.T) { require.False(t, srv.ShouldAccelerateBlock(110), "failed poll must not clear the list") require.True(t, srv.ShouldAccelerateBlock(100)) } + +// Without a prime the window stays empty until the first 30s tick, so a restarted +// gateway accelerates every block for half a minute: fail-open for want of an +// answer rather than because one was unavailable. +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) + + // Slot 110 is inside the horizon and unselected, so it only stops accelerating + // once the window has been fetched. The prime runs on bgSync's goroutine. + 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") +} diff --git a/pkg/service/message_router/bg_sync.go b/pkg/service/message_router/bg_sync.go index 3deab83..ecfff31 100644 --- a/pkg/service/message_router/bg_sync.go +++ b/pkg/service/message_router/bg_sync.go @@ -16,6 +16,11 @@ 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 before the first tick: otherwise every restart accelerates every block + // for 30s, which is the fail-open path taken for want of an answer rather than + // because one was unavailable. + s.RefreshAccelerateSlots(ctx) + ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { diff --git a/pkg/test_utils/local_bootstrap_server.go b/pkg/test_utils/local_bootstrap_server.go index 7cce560..ff5f703 100644 --- a/pkg/test_utils/local_bootstrap_server.go +++ b/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,17 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ } } +// requireAuth answers 401 rather than asserting. These run on fiber's handler +// goroutine, where a failed require calls t.FailNow off the test goroutine: Go +// does not support that, and the caller sees a transport error instead of the +// missing header. A real status lets the caller report it. +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 { From 46ed364f53d45e2ad12953bf6b0301e884aab109 Mon Sep 17 00:00:00 2001 From: Chris Fuka Date: Mon, 31 Aug 2026 14:11:55 -0500 Subject: [PATCH 2/2] fix(message-router): bound the token mint, and keep the prime off the network Review found the prime had made two latent things live. Token mint. Moving ServicesToken outside the poll deadline left it on the process context, and mint runs RetryPostRequest on http.DefaultClient, which has no timeout. Since the prime is synchronous at the top of bgSync, a hung auth would stall SetKnownValidators as well, forever, in exactly the cold-cache case the prime exists for. Verified: no deadline hangs past 120s, a 500ms one returns in 500ms. Mint now takes its own 5s, then the poll takes its own. newTestService pointed at dev-bootstrap.getoptimum.io. That was inert while nothing in bgSync did I/O; with the prime every caller fired a real request, carrying a rig-signed JWT, at a shared host. It stubs 404s locally now. The gate test seeded its window after constructing the router, so the prime read an empty stub and could store fail-open after the explicit refresh. newGateway takes prepare hooks that run before any service polls. Also: log the ServicesToken error instead of discarding it, since a failed mint otherwise surfaces only as a 401 that reads as a bootstrap fault; hoist the verdict strings into constants, they were duplicated between decideAccelerate and the gate; swap in-handler require for assert, the same FailNow-off-the-test- goroutine problem this branch already fixed for the fiber handlers. Comments trimmed to one line throughout, per review. --- .../gossipsub-gateway/accelerate_gate_test.go | 16 +++++----- pkg/service/gossipsub-gateway/setup_test.go | 6 +++- pkg/service/message_router/accelerate.go | 30 ++++++++++++------- pkg/service/message_router/accelerate_test.go | 15 ++++------ pkg/service/message_router/bg_sync.go | 4 +-- pkg/service/message_router/service_test.go | 9 +++++- pkg/test_utils/local_bootstrap_server.go | 5 +--- 7 files changed, 49 insertions(+), 36 deletions(-) diff --git a/pkg/service/gossipsub-gateway/accelerate_gate_test.go b/pkg/service/gossipsub-gateway/accelerate_gate_test.go index f8c4e61..b3996dc 100644 --- a/pkg/service/gossipsub-gateway/accelerate_gate_test.go +++ b/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 diff --git a/pkg/service/gossipsub-gateway/setup_test.go b/pkg/service/gossipsub-gateway/setup_test.go index c3741a2..e316d38 100644 --- a/pkg/service/gossipsub-gateway/setup_test.go +++ b/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) diff --git a/pkg/service/message_router/accelerate.go b/pkg/service/message_router/accelerate.go index 710f8aa..9028aa6 100644 --- a/pkg/service/message_router/accelerate.go +++ b/pkg/service/message_router/accelerate.go @@ -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,18 +53,21 @@ func (s *Service) RefreshAccelerateSlots(ctx context.Context) { if chainID == "" || s.cfg.RemoteBootstrapURL == "" { return } - // Token first, outside the deadline below: ServicesToken mints over HTTP when - // nothing is cached, and that must not eat the poll's budget. + // 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) - // status_code is a field rather than part of err: a bootstrap without the - // endpoint serves a non-JSON 404, which GetCurl reports as an unmarshal error - // alongside the code, and the code is the half that says what is wrong. + // 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 diff --git a/pkg/service/message_router/accelerate_test.go b/pkg/service/message_router/accelerate_test.go index fea1626..b06d232 100644 --- a/pkg/service/message_router/accelerate_test.go +++ b/pkg/service/message_router/accelerate_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" commonentities "github.com/getoptimum/optimum-common/pkg/entities" @@ -15,12 +16,11 @@ import ( func TestShouldAccelerateBlock(t *testing.T) { var fail atomic.Bool - // bgSync primes on startup, so serve failures until the fail-open assertion - // below is done. A failed poll leaves the window nil, which is what it needs. - fail.Store(true) + 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 @@ -49,9 +49,6 @@ func TestShouldAccelerateBlock(t *testing.T) { require.True(t, srv.ShouldAccelerateBlock(100)) } -// Without a prime the window stays empty until the first 30s tick, so a restarted -// gateway accelerates every block for half a minute: fail-open for want of an -// answer rather than because one was unavailable. func TestAccelerateSlotsPrimedAtStartup(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _ = json.NewEncoder(w).Encode(map[string]any{ @@ -64,8 +61,6 @@ func TestAccelerateSlotsPrimedAtStartup(t *testing.T) { srv := newTestServiceAt(t, commonentities.GatewayTypePartner, ts.URL) - // Slot 110 is inside the horizon and unselected, so it only stops accelerating - // once the window has been fetched. The prime runs on bgSync's goroutine. 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") diff --git a/pkg/service/message_router/bg_sync.go b/pkg/service/message_router/bg_sync.go index ecfff31..9b97c17 100644 --- a/pkg/service/message_router/bg_sync.go +++ b/pkg/service/message_router/bg_sync.go @@ -16,9 +16,7 @@ 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 before the first tick: otherwise every restart accelerates every block - // for 30s, which is the fail-open path taken for want of an answer rather than - // because one was unavailable. + // Prime so a restart is not fail-open until the first 30s tick. s.RefreshAccelerateSlots(ctx) ticker := time.NewTicker(30 * time.Second) diff --git a/pkg/service/message_router/service_test.go b/pkg/service/message_router/service_test.go index 05be59a..314c47d 100644 --- a/pkg/service/message_router/service_test.go +++ b/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 { diff --git a/pkg/test_utils/local_bootstrap_server.go b/pkg/test_utils/local_bootstrap_server.go index ff5f703..e083394 100644 --- a/pkg/test_utils/local_bootstrap_server.go +++ b/pkg/test_utils/local_bootstrap_server.go @@ -125,10 +125,7 @@ func newLocalBootstrapServer(t *testing.T, rig *AuthTestRig) *LocalBootstrapServ } } -// requireAuth answers 401 rather than asserting. These run on fiber's handler -// goroutine, where a failed require calls t.FailNow off the test goroutine: Go -// does not support that, and the caller sees a transport error instead of the -// missing header. A real status lets the caller report it. +// Fiber handlers cannot require.FailNow; return 401 instead. func requireAuth(rig *AuthTestRig, c fiber.Ctx) error { if rig == nil || c.HasHeader("Authorization") { return nil