feat: poll bootstrap accelerate_slots and gateway beacon blocks #96
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| 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) { | ||
| 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() | ||
| svc.streamHub = hub | ||
| sub := hub.Subscribe(4) | ||
| t.Cleanup(sub.Close) | ||
| t.Cleanup(svc.messagesMap.Close) | ||
|
|
||
| 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") | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| package message_router | ||
|
|
||
| import ( | ||
| "context" | ||
| "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" | ||
| ) | ||
|
|
||
| type accelerateWindow struct { | ||
| toSlot uint64 | ||
| slots map[uint64]struct{} | ||
| } | ||
|
|
||
| type accelerateSlotsResponse struct { | ||
| ToSlot int64 `json:"to_slot"` | ||
| Slots []int64 `json:"slots"` | ||
| 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 != accelerateNotOnList | ||
| } | ||
|
|
||
| func decideAccelerate(w *accelerateWindow, slot uint64) string { | ||
| if w == nil || w.toSlot == 0 || slot > w.toSlot { | ||
| return accelerateFailOpen | ||
|
Comment on lines
+41
to
+42
There was a problem hiding this comment. 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win Preserve fail-open behavior before the examined window. Bootstrap examines only As per coding guidelines, "Flag changed paths that may silently alter semantics or break invariants. Require focused tests for non-trivial behavior changes." 🤖 Prompt for AI AgentsSources: Coding guidelines, Path instructions |
||
| } | ||
| if _, ok := w.slots[slot]; ok { | ||
| return accelerateOnList | ||
| } | ||
| return accelerateNotOnList | ||
| } | ||
|
|
||
| // 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 | ||
| } | ||
| // Mint gets its own deadline: it runs on http.DefaultClient, so a hung auth stalls bgSync. | ||
| var headers map[string]string | ||
| 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) | ||
| // 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))} | ||
| if res.ToSlot > 0 { | ||
| w.toSlot = uint64(res.ToSlot) | ||
|
Comment on lines
+75
to
+77
There was a problem hiding this comment. 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win 🔎 Supported by static analysis🏁 Script executed: #!/bin/bash
set -euo pipefail
rg -n -C 4 --glob '*.go' --glob '*.md' --glob '*.json' --glob '*.yaml' \
'accelerate_slots|generated_at_ms|to_slot|slots' .Repository: getoptimum/optimum-gateway Length of output: 50384 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- applicable repository guidance ---'
head -5 /tmp/coderabbit-repo-knowledge/getoptimum-optimum-gateway-a2b26dfa/*/*.md 2>/dev/null || true
printf '%s\n' '--- changed file and adjacent tests ---'
sed -n '1,130p' pkg/service/message_router/accelerate.go
sed -n '1,120p' pkg/service/message_router/accelerate_test.go
printf '%s\n' '--- ADR contract around refresh and failure modes ---'
sed -n '70,115p' docs/adr/0012-slot-based-block-acceleration.md
printf '%s\n' '--- focused diff ---'
git diff -- pkg/service/message_router/accelerate.go pkg/service/message_router/accelerate_test.goRepository: getoptimum/optimum-gateway Length of output: 8701 Reject a missing If a 200 response omits 🤖 Prompt for AI AgentsSources: Coding guidelines, Path instructions |
||
| } | ||
| for _, slot := range res.Slots { | ||
| if slot >= 0 { | ||
| w.slots[uint64(slot)] = struct{}{} | ||
| } | ||
| } | ||
| s.accelerate.Store(w) | ||
| telemetry.SetAccelerateWindow(w.toSlot, res.GeneratedAtMs) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package message_router_test | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "net/http" | ||
| "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) { | ||
| // 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 | ||
| } | ||
| _ = json.NewEncoder(w).Encode(map[string]any{ | ||
| "to_slot": 120, | ||
| "slots": []int64{100, 101}, | ||
| "generated_at_ms": 1, | ||
| }) | ||
| })) | ||
| t.Cleanup(ts.Close) | ||
|
|
||
| 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)) | ||
| require.False(t, srv.ShouldAccelerateBlock(110), "examined, not selected") | ||
| require.True(t, srv.ShouldAccelerateBlock(121), "past to_slot fail-opens") | ||
|
|
||
| fail.Store(true) | ||
| srv.RefreshAccelerateSlots(t.Context()) | ||
| 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") | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add focused tests for both forwarding gates.
These lines change whether beacon blocks reach Mump2P and the CL topic. Add observable tests that load a selected and an unselected slot, then verify that each inbound path publishes only the selected block. Use the existing gateway test fixture and assert publish results, not internal state.
As per coding guidelines, “Require focused tests for non-trivial behavior changes.” As per path instructions, “Prefer focused tests on changed behavior only.”
Also applies to: 145-147
🤖 Prompt for AI Agents
Sources: Coding guidelines, Path instructions