diff --git a/pkg/config/config.go b/pkg/config/config.go index 73409d5..b0d2215 100644 --- a/pkg/config/config.go +++ b/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) } diff --git a/pkg/service/gossipsub-gateway/accelerate_gate_test.go b/pkg/service/gossipsub-gateway/accelerate_gate_test.go new file mode 100644 index 0000000..f8c4e61 --- /dev/null +++ b/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") +} diff --git a/pkg/service/message_router/accelerate.go b/pkg/service/message_router/accelerate.go index 7459afb..a64bb84 100644 --- a/pkg/service/message_router/accelerate.go +++ b/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 diff --git a/pkg/service/message_router/accelerate_test.go b/pkg/service/message_router/accelerate_test.go index 74258c1..0674aff 100644 --- a/pkg/service/message_router/accelerate_test.go +++ b/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)) } diff --git a/pkg/service/message_router/bg_sync.go b/pkg/service/message_router/bg_sync.go index 5f4a491..3deab83 100644 --- a/pkg/service/message_router/bg_sync.go +++ b/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) } } } diff --git a/pkg/service/message_router/export_test.go b/pkg/service/message_router/export_test.go deleted file mode 100644 index 1594ec3..0000000 --- a/pkg/service/message_router/export_test.go +++ /dev/null @@ -1,10 +0,0 @@ -package message_router - -import "testing" - -// PollAccelerateSlotsForTest points the router at bootstrapURL and runs one poll. -func PollAccelerateSlotsForTest(t *testing.T, s *Service, bootstrapURL string) { - t.Helper() - s.cfg.RemoteBootstrapURL = bootstrapURL - s.pollAccelerateSlots(t.Context()) -} diff --git a/pkg/service/message_router/service_test.go b/pkg/service/message_router/service_test.go index b41d0c4..05be59a 100644 --- a/pkg/service/message_router/service_test.go +++ b/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) diff --git a/pkg/test_utils/jwt_auth_claims.go b/pkg/test_utils/jwt_auth_claims.go index cc74f8c..570700d 100644 --- a/pkg/test_utils/jwt_auth_claims.go +++ b/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 } diff --git a/pkg/test_utils/local_bootstrap_server.go b/pkg/test_utils/local_bootstrap_server.go index b9ac68e..7cce560 100644 --- a/pkg/test_utils/local_bootstrap_server.go +++ b/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) }