From ff2b973a4771237549e4fbe926eee77f2293fbcd Mon Sep 17 00:00:00 2001 From: AntiD2ta Date: Wed, 25 Mar 2026 17:47:22 +0100 Subject: [PATCH 1/3] Add provider address to event data structs. Allow consumers to identify which beacon node delivered each event. --- api/v1/blobsidecarevent.go | 2 + api/v1/blobsidecarevent_test.go | 17 +++ api/v1/blockevent.go | 2 + api/v1/blockevent_test.go | 17 +++ api/v1/blockgossipevent.go | 2 + api/v1/blockgossipevent_test.go | 17 +++ api/v1/chainreorgevent.go | 2 + api/v1/chainreorgevent_test.go | 17 +++ api/v1/datacolumnsidecarevent.go | 2 + api/v1/datacolumnsidecarevent_test.go | 17 +++ api/v1/event.go | 2 + api/v1/event_test.go | 17 +++ api/v1/finalizedcheckpointevent.go | 2 + api/v1/finalizedcheckpointevent_test.go | 17 +++ api/v1/headevent.go | 2 + api/v1/headevent_test.go | 17 +++ api/v1/payloadattributesevent.go | 2 + api/v1/payloadattributesevent_test.go | 17 +++ api/v1/provider.go | 37 ++++++ api/v1/provider_test.go | 62 ++++++++++ http/events.go | 143 ++++++++++++++++-------- multi/events.go | 18 +++ 22 files changed, 386 insertions(+), 45 deletions(-) create mode 100644 api/v1/provider.go create mode 100644 api/v1/provider_test.go diff --git a/api/v1/blobsidecarevent.go b/api/v1/blobsidecarevent.go index 28805a916..811dc1973 100644 --- a/api/v1/blobsidecarevent.go +++ b/api/v1/blobsidecarevent.go @@ -29,6 +29,8 @@ type BlobSidecarEvent struct { Index deneb.BlobIndex KZGCommitment deneb.KZGCommitment VersionedHash deneb.VersionedHash + // Provider is the address of the provider that returned this event. + Provider string } // blobSidecarEventJSON is the spec representation of the struct. diff --git a/api/v1/blobsidecarevent_test.go b/api/v1/blobsidecarevent_test.go index dc412ea12..bfded0d3f 100644 --- a/api/v1/blobsidecarevent_test.go +++ b/api/v1/blobsidecarevent_test.go @@ -151,3 +151,20 @@ func TestBlobSidecarEventJSON(t *testing.T) { }) } } + +func TestBlobSidecarEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"block_root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2","slot":"1","index":"1","kzg_commitment":"0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2","versioned_hash":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2"}`) + + var event api.BlobSidecarEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.BlobSidecarEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/blockevent.go b/api/v1/blockevent.go index 0b54dd8b3..ae6701b6e 100644 --- a/api/v1/blockevent.go +++ b/api/v1/blockevent.go @@ -29,6 +29,8 @@ type BlockEvent struct { Slot phase0.Slot Block phase0.Root ExecutionOptimistic bool + // Provider is the address of the provider that returned this event. + Provider string } // blockEventJSON is the spec representation of the struct. diff --git a/api/v1/blockevent_test.go b/api/v1/blockevent_test.go index 42d8760f3..7048e596c 100644 --- a/api/v1/blockevent_test.go +++ b/api/v1/blockevent_test.go @@ -108,3 +108,20 @@ func TestBlockEventJSON(t *testing.T) { }) } } + +func TestBlockEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"slot":"525277","block":"0x99e3f24aab3dd084045a0c927a33b8463eb5c7b17eeadfecdcf4e4badf7b6028","execution_optimistic":false}`) + + var event api.BlockEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.BlockEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/blockgossipevent.go b/api/v1/blockgossipevent.go index 0f4be750f..93171c550 100644 --- a/api/v1/blockgossipevent.go +++ b/api/v1/blockgossipevent.go @@ -28,6 +28,8 @@ import ( type BlockGossipEvent struct { Slot phase0.Slot Block phase0.Root + // Provider is the address of the provider that returned this event. + Provider string } // blockGossipEventJSON is the spec representation of the struct. diff --git a/api/v1/blockgossipevent_test.go b/api/v1/blockgossipevent_test.go index b73fae73a..268a58a93 100644 --- a/api/v1/blockgossipevent_test.go +++ b/api/v1/blockgossipevent_test.go @@ -99,3 +99,20 @@ func TestBlockGossipEventJSON(t *testing.T) { }) } } + +func TestBlockGossipEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"slot":"525277","block":"0x99e3f24aab3dd084045a0c927a33b8463eb5c7b17eeadfecdcf4e4badf7b6028"}`) + + var event api.BlockGossipEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.BlockGossipEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/chainreorgevent.go b/api/v1/chainreorgevent.go index efa76e0a2..264bf6d6f 100644 --- a/api/v1/chainreorgevent.go +++ b/api/v1/chainreorgevent.go @@ -33,6 +33,8 @@ type ChainReorgEvent struct { OldHeadState phase0.Root NewHeadState phase0.Root Epoch phase0.Epoch + // Provider is the address of the provider that returned this event. + Provider string } // chainReorgEventJSON is the spec representation of the struct. diff --git a/api/v1/chainreorgevent_test.go b/api/v1/chainreorgevent_test.go index c5760393e..165078c78 100644 --- a/api/v1/chainreorgevent_test.go +++ b/api/v1/chainreorgevent_test.go @@ -204,3 +204,20 @@ func TestChainReorgEventJSON(t *testing.T) { }) } } + +func TestChainReorgEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"slot":"524986","depth":"2","old_head_block":"0x2ffc0a5b75de20f2a12853dff3e09b263e7c3cb19515134cba756b28e5ba25ee","new_head_block":"0xa3fe14d8d749318359aa3790d3588a23e12ea3b02bd879fbfbf04c3a66770df7","old_head_state":"0x97cc0a37b77fbac6fa140f330c92521ddcd5b1dfefeef99d86996a51f1993d60","new_head_state":"0x4ab800aaa51c14c786fe7e924abd1355aa2ac2e0434d7cb5ae568720ed1bf522","epoch":"16405"}`) + + var event api.ChainReorgEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.ChainReorgEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/datacolumnsidecarevent.go b/api/v1/datacolumnsidecarevent.go index aa69f21df..d7aad39e3 100644 --- a/api/v1/datacolumnsidecarevent.go +++ b/api/v1/datacolumnsidecarevent.go @@ -28,6 +28,8 @@ type DataColumnSidecarEvent struct { Slot phase0.Slot Index uint64 KZGCommitments []deneb.KZGCommitment + // Provider is the address of the provider that returned this event. + Provider string } // dataColumnSidecarEventJSON is the spec representation of the struct. diff --git a/api/v1/datacolumnsidecarevent_test.go b/api/v1/datacolumnsidecarevent_test.go index 921a09f2a..1956a3508 100644 --- a/api/v1/datacolumnsidecarevent_test.go +++ b/api/v1/datacolumnsidecarevent_test.go @@ -103,3 +103,20 @@ func TestDataColumnSidecarEventJSON(t *testing.T) { }) } } + +func TestDataColumnSidecarEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"block_root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2","slot":"1","index":"1","kzg_commitments":["0xa590e760fdce951756d59c46b037bab8de815fe8ffc25e6e3a7b45e43289e1fdc942854cdfea1615385a0db63442f363"]}`) + + var event api.DataColumnSidecarEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.DataColumnSidecarEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/event.go b/api/v1/event.go index cb271b0d4..0d949e313 100644 --- a/api/v1/event.go +++ b/api/v1/event.go @@ -31,6 +31,8 @@ type Event struct { Topic string // Data is the data of the event. Data any + // Provider is the address of the provider that returned this event. + Provider string } // SupportedEventTopics is a map of supported event topics. diff --git a/api/v1/event_test.go b/api/v1/event_test.go index c225e1e4b..6c64672e8 100644 --- a/api/v1/event_test.go +++ b/api/v1/event_test.go @@ -111,3 +111,20 @@ func TestEvent(t *testing.T) { }) } } + +func TestEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"topic":"head","data":{"block":"0xbe36e714a6114cf718e35dafc4ac530ce8f01e4a9a360e78098eb129772dcc39","current_duty_dependent_root":"0x92c6b763f610d5941d2041906007bf9449d37772aacf0483a76275ac27c096b4","epoch_transition":false,"previous_duty_dependent_root":"0xa692c095bbca3eeaf99eeabada78874c028c02b176ccf691f3e8fa075d67f5c6","slot":"231192","state":"0x61099b2c1dee0104c93ce0e14e5f5fc4b6faceff4cb863278d055bdfb73b7dc7"}}`) + + var event api.Event + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "\"provider\"") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.Event + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/finalizedcheckpointevent.go b/api/v1/finalizedcheckpointevent.go index 471d8a159..19cf23f0e 100644 --- a/api/v1/finalizedcheckpointevent.go +++ b/api/v1/finalizedcheckpointevent.go @@ -29,6 +29,8 @@ type FinalizedCheckpointEvent struct { Block phase0.Root State phase0.Root Epoch phase0.Epoch + // Provider is the address of the provider that returned this event. + Provider string } // finalizedCheckpointEventJSON is the spec representation of the struct. diff --git a/api/v1/finalizedcheckpointevent_test.go b/api/v1/finalizedcheckpointevent_test.go index 98a7d0184..a8f18bfa5 100644 --- a/api/v1/finalizedcheckpointevent_test.go +++ b/api/v1/finalizedcheckpointevent_test.go @@ -124,3 +124,20 @@ func TestFinalizedCheckpointEventJSON(t *testing.T) { }) } } + +func TestFinalizedCheckpointEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"block":"0x99e3f24aab3dd084045a0c927a33b8463eb5c7b17eeadfecdcf4e4badf7b6028","state":"0x749a95b1355828b758864ea601c007e69aabed7b34a0f2084c43c26242f77e28","epoch":"2"}`) + + var event api.FinalizedCheckpointEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.FinalizedCheckpointEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/headevent.go b/api/v1/headevent.go index 8d5894ed7..d6f4124eb 100644 --- a/api/v1/headevent.go +++ b/api/v1/headevent.go @@ -33,6 +33,8 @@ type HeadEvent struct { EpochTransition bool CurrentDutyDependentRoot phase0.Root PreviousDutyDependentRoot phase0.Root + // Provider is the address of the provider that returned this event. + Provider string } // headEventJSON is the spec representation of the struct. diff --git a/api/v1/headevent_test.go b/api/v1/headevent_test.go index 208d4f01c..3f62fa7a3 100644 --- a/api/v1/headevent_test.go +++ b/api/v1/headevent_test.go @@ -177,3 +177,20 @@ func TestHeadEventJSON(t *testing.T) { }) } } + +func TestHeadEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"slot":"525277","block":"0x99e3f24aab3dd084045a0c927a33b8463eb5c7b17eeadfecdcf4e4badf7b6028","state":"0x749a95b1355828b758864ea601c007e69aabed7b34a0f2084c43c26242f77e28","epoch_transition":false,"current_duty_dependent_root":"0x907a3462a2905e3df2624869aa7f9a8635eb35bdcf9ce68a26fab691f9dada61","previous_duty_dependent_root":"0x935569bdc1aaad65dbeb532a125390d039058924ea81799238ed53e4e4639a11"}`) + + var event api.HeadEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "provider") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.HeadEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/payloadattributesevent.go b/api/v1/payloadattributesevent.go index 93a19bedb..7542239d2 100644 --- a/api/v1/payloadattributesevent.go +++ b/api/v1/payloadattributesevent.go @@ -34,6 +34,8 @@ type PayloadAttributesEvent struct { Version spec.DataVersion // Data is the data of the event. Data *PayloadAttributesData + // Provider is the address of the provider that returned this event. + Provider string } // PayloadAttributesData represents the data of a payload_attributes event. diff --git a/api/v1/payloadattributesevent_test.go b/api/v1/payloadattributesevent_test.go index 3c18445a3..2696c51c4 100644 --- a/api/v1/payloadattributesevent_test.go +++ b/api/v1/payloadattributesevent_test.go @@ -205,3 +205,20 @@ func TestPayloadAttributesEventJSON(t *testing.T) { }) } } + +func TestPayloadAttributesEventProviderNotInJSON(t *testing.T) { + input := []byte(`{"version":"bellatrix","data":{"proposer_index":"123","proposal_slot":"10","parent_block_number":"9","parent_block_root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2","parent_block_hash":"0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf","payload_attributes":{"timestamp":"123456","prev_randao":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2","suggested_fee_recipient":"0x0000000000000000000000000000000000000000"}}}`) + + var event api.PayloadAttributesEvent + require.NoError(t, json.Unmarshal(input, &event)) + event.Provider = "test-provider" + + rt, err := json.Marshal(&event) + require.NoError(t, err) + assert.NotContains(t, string(rt), "\"provider\"") + assert.NotContains(t, string(rt), "test-provider") + + var roundTripped api.PayloadAttributesEvent + require.NoError(t, json.Unmarshal(rt, &roundTripped)) + assert.Empty(t, roundTripped.Provider) +} diff --git a/api/v1/provider.go b/api/v1/provider.go new file mode 100644 index 000000000..8f5eb2365 --- /dev/null +++ b/api/v1/provider.go @@ -0,0 +1,37 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package v1 + +import "context" + +type providerKey struct{} + +// WithProvider returns a context with the provider address set. +// This is the universal mechanism for obtaining the provider address +// across all event types. Event structs owned by this package also +// carry the address directly in their Provider field. +func WithProvider(ctx context.Context, provider string) context.Context { + return context.WithValue(ctx, providerKey{}, provider) +} + +// Provider extracts the provider address from the context. +// It returns an empty string if no provider has been set. +func Provider(ctx context.Context) string { + provider, ok := ctx.Value(providerKey{}).(string) + if !ok { + return "" + } + + return provider +} diff --git a/api/v1/provider_test.go b/api/v1/provider_test.go new file mode 100644 index 000000000..7cee1df42 --- /dev/null +++ b/api/v1/provider_test.go @@ -0,0 +1,62 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package v1_test + +import ( + "context" + "testing" + + v1 "github.com/attestantio/go-eth2-client/api/v1" + require "github.com/stretchr/testify/require" +) + +func TestProvider(t *testing.T) { + tests := []struct { + name string + setup func(ctx context.Context) context.Context + expected string + }{ + { + name: "Empty", + setup: func(ctx context.Context) context.Context { + return ctx + }, + expected: "", + }, + { + name: "Set", + setup: func(ctx context.Context) context.Context { + return v1.WithProvider(ctx, "http://localhost:5052") + }, + expected: "http://localhost:5052", + }, + { + name: "Override", + setup: func(ctx context.Context) context.Context { + ctx = v1.WithProvider(ctx, "http://localhost:5052") + return v1.WithProvider(ctx, "http://localhost:5053") + }, + expected: "http://localhost:5053", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctx := context.Background() + ctx = test.setup(ctx) + result := v1.Provider(ctx) + require.Equal(t, test.expected, result) + }) + } +} diff --git a/http/events.go b/http/events.go index 6d8067f2b..a9c17e120 100644 --- a/http/events.go +++ b/http/events.go @@ -219,7 +219,7 @@ func (s *Service) handleEvent(ctx context.Context, } } -func (*Service) handleAttestationEvent(ctx context.Context, +func (s *Service) handleAttestationEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -233,20 +233,23 @@ func (*Service) handleAttestationEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.AttestationHandler != nil: opts.AttestationHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleAttesterSlashingEvent(ctx context.Context, +func (s *Service) handleAttesterSlashingEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -260,20 +263,23 @@ func (*Service) handleAttesterSlashingEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.AttesterSlashingHandler != nil: opts.AttesterSlashingHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleBlobSidecarEvent(ctx context.Context, +func (s *Service) handleBlobSidecarEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -287,20 +293,24 @@ func (*Service) handleBlobSidecarEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.BlobSidecarHandler != nil: opts.BlobSidecarHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleBlockEvent(ctx context.Context, +func (s *Service) handleBlockEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -314,20 +324,24 @@ func (*Service) handleBlockEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.BlockHandler != nil: opts.BlockHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleBlockGossipEvent(ctx context.Context, +func (s *Service) handleBlockGossipEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -341,20 +355,24 @@ func (*Service) handleBlockGossipEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.BlockGossipHandler != nil: opts.BlockGossipHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleBLSToExecutionChangeEvent(ctx context.Context, +func (s *Service) handleBLSToExecutionChangeEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -368,20 +386,23 @@ func (*Service) handleBLSToExecutionChangeEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.BLSToExecutionChangeHandler != nil: opts.BLSToExecutionChangeHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleChainReorgEvent(ctx context.Context, +func (s *Service) handleChainReorgEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -395,20 +416,24 @@ func (*Service) handleChainReorgEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.ChainReorgHandler != nil: opts.ChainReorgHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleContributionAndProofEvent(ctx context.Context, +func (s *Service) handleContributionAndProofEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -422,20 +447,23 @@ func (*Service) handleContributionAndProofEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.ContributionAndProofHandler != nil: opts.ContributionAndProofHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleFinalizedCheckpointEvent(ctx context.Context, +func (s *Service) handleFinalizedCheckpointEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -449,20 +477,24 @@ func (*Service) handleFinalizedCheckpointEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.FinalizedCheckpointHandler != nil: opts.FinalizedCheckpointHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleHeadEvent(ctx context.Context, +func (s *Service) handleHeadEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -476,20 +508,24 @@ func (*Service) handleHeadEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.HeadHandler != nil: opts.HeadHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handlePayloadAttributesEvent(ctx context.Context, +func (s *Service) handlePayloadAttributesEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -503,20 +539,24 @@ func (*Service) handlePayloadAttributesEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.PayloadAttributesHandler != nil: opts.PayloadAttributesHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleProposerSlashingEvent(ctx context.Context, +func (s *Service) handleProposerSlashingEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -530,20 +570,23 @@ func (*Service) handleProposerSlashingEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.ProposerSlashingHandler != nil: opts.ProposerSlashingHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleSingleAttestationEvent(ctx context.Context, +func (s *Service) handleSingleAttestationEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -557,20 +600,23 @@ func (*Service) handleSingleAttestationEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.SingleAttestationHandler != nil: opts.SingleAttestationHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleVoluntaryExitEvent(ctx context.Context, +func (s *Service) handleVoluntaryExitEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -584,20 +630,23 @@ func (*Service) handleVoluntaryExitEvent(ctx context.Context, return } + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.VoluntaryExitHandler != nil: opts.VoluntaryExitHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") } } -func (*Service) handleDataColumnSidecarEvent(ctx context.Context, +func (s *Service) handleDataColumnSidecarEvent(ctx context.Context, msg *sse.Event, opts *api.EventsOpts, ) { @@ -611,13 +660,17 @@ func (*Service) handleDataColumnSidecarEvent(ctx context.Context, return } + data.Provider = s.address + ctx = apiv1.WithProvider(ctx, s.address) + switch { case opts.DataColumnSidecarHandler != nil: opts.DataColumnSidecarHandler(ctx, data) case opts.Handler != nil: opts.Handler(&apiv1.Event{ - Topic: string(msg.Event), - Data: data, + Topic: string(msg.Event), + Data: data, + Provider: s.address, }) default: log.Debug().Msg("No specific or generic handler supplied; ignoring") diff --git a/multi/events.go b/multi/events.go index 4b75ad15e..6103cb1fe 100644 --- a/multi/events.go +++ b/multi/events.go @@ -156,6 +156,7 @@ func (h *activeHandler) attestationHandler(ctx context.Context, data *spec.Versi log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.AttestationHandler(ctx, data) } @@ -172,6 +173,7 @@ func (h *activeHandler) attesterSlashingHandler(ctx context.Context, data *elect log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.AttesterSlashingHandler(ctx, data) } @@ -188,6 +190,8 @@ func (h *activeHandler) blobSidecarHandler(ctx context.Context, data *apiv1.Blob log.Trace().Msg("Forwarding due to primary active address") + data.Provider = h.address + ctx = apiv1.WithProvider(ctx, h.address) h.opts.BlobSidecarHandler(ctx, data) } @@ -204,6 +208,7 @@ func (h *activeHandler) blsToExecutionChangeHandler(ctx context.Context, data *c log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.BLSToExecutionChangeHandler(ctx, data) } @@ -220,6 +225,8 @@ func (h *activeHandler) chainReorgHandler(ctx context.Context, data *apiv1.Chain log.Trace().Msg("Forwarding due to primary active address") + data.Provider = h.address + ctx = apiv1.WithProvider(ctx, h.address) h.opts.ChainReorgHandler(ctx, data) } @@ -236,6 +243,7 @@ func (h *activeHandler) contributionAndProofHandler(ctx context.Context, data *a log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.ContributionAndProofHandler(ctx, data) } @@ -252,6 +260,8 @@ func (h *activeHandler) finalizedCheckpointHandler(ctx context.Context, data *ap log.Trace().Msg("Forwarding due to primary active address") + data.Provider = h.address + ctx = apiv1.WithProvider(ctx, h.address) h.opts.FinalizedCheckpointHandler(ctx, data) } @@ -268,6 +278,8 @@ func (h *activeHandler) headHandler(ctx context.Context, data *apiv1.HeadEvent) log.Trace().Msg("Forwarding due to primary active address") + data.Provider = h.address + ctx = apiv1.WithProvider(ctx, h.address) h.opts.HeadHandler(ctx, data) } @@ -284,6 +296,8 @@ func (h *activeHandler) payloadAttributesHandler(ctx context.Context, data *apiv log.Trace().Msg("Forwarding due to primary active address") + data.Provider = h.address + ctx = apiv1.WithProvider(ctx, h.address) h.opts.PayloadAttributesHandler(ctx, data) } @@ -300,6 +314,7 @@ func (h *activeHandler) proposerSlashingHandler(ctx context.Context, data *phase log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.ProposerSlashingHandler(ctx, data) } @@ -316,6 +331,7 @@ func (h *activeHandler) singleAttestationHandler(ctx context.Context, data *elec log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.SingleAttestationHandler(ctx, data) } @@ -332,6 +348,7 @@ func (h *activeHandler) voluntaryExitHandler(ctx context.Context, data *phase0.S log.Trace().Msg("Forwarding due to primary active address") + ctx = apiv1.WithProvider(ctx, h.address) h.opts.VoluntaryExitHandler(ctx, data) } @@ -348,6 +365,7 @@ func (h *activeHandler) genericHandler(event *apiv1.Event) { log.Trace().Msg("Forwarding due to primary active address") + event.Provider = h.address if h.opts.Handler != nil { h.opts.Handler(event) } From d5ed32d841c25f01f3b408920bce19fb6cad5d92 Mon Sep 17 00:00:00 2001 From: AntiD2ta Date: Wed, 25 Mar 2026 17:54:20 +0100 Subject: [PATCH 2/3] Remove unused nolint directives. --- api/v1/electra/generate.go | 2 +- spec/fulu/generate.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/api/v1/electra/generate.go b/api/v1/electra/generate.go index a4d288cf1..dc8647c45 100644 --- a/api/v1/electra/generate.go +++ b/api/v1/electra/generate.go @@ -13,8 +13,8 @@ package electra -//nolint:revive // Need to `go install github.com/ferranbt/fastssz/sszgen@latest` for this to work. +//nolint:revive // go:generate directives exceed line length. //go:generate rm -f blindedbeaconblock_ssz.go blindedbeaconblockbody_ssz.go blockcontents_ssz.go signedblindedbeaconblock_ssz.go signedblockcontents_ssz.go //go:generate sszgen --include ../../../spec/phase0,../../../spec/altair,../../../spec/bellatrix,../../../spec/capella,../../../spec/deneb,../../../spec/electra -path . --suffix ssz -objs BlindedBeaconBlock,BlindedBeaconBlockBody,BlockContents,SignedBlindedBeaconBlock,SignedBlockContents //go:generate goimports -w blindedbeaconblock_ssz.go blindedbeaconblockbody_ssz.go blockcontents_ssz.go signedblindedbeaconblock_ssz.go signedblockcontents_ssz.go diff --git a/spec/fulu/generate.go b/spec/fulu/generate.go index b0d243101..18eff8127 100644 --- a/spec/fulu/generate.go +++ b/spec/fulu/generate.go @@ -15,6 +15,6 @@ package fulu // Need to `go install github.com/ferranbt/fastssz/sszgen@latest` for this to work. //go:generate rm -f beaconstate_ssz.go -//nolint:revive +//nolint:revive // go:generate directive exceeds line length. //go:generate sszgen --suffix=ssz --path . --include ../phase0,../altair,../bellatrix,../capella,../deneb,../electra --objs BeaconState //go:generate goimports -w beaconstate_ssz.go From 346dc296d0150ec114c7537f34133a320b3aedf6 Mon Sep 17 00:00:00 2001 From: AntiD2ta Date: Thu, 26 Mar 2026 15:34:26 +0100 Subject: [PATCH 3/3] Remove context-based provider; keep Provider field on apiv1 event structs --- api/v1/provider.go | 37 ------------------------ api/v1/provider_test.go | 62 ----------------------------------------- http/events.go | 22 --------------- multi/events.go | 12 -------- 4 files changed, 133 deletions(-) delete mode 100644 api/v1/provider.go delete mode 100644 api/v1/provider_test.go diff --git a/api/v1/provider.go b/api/v1/provider.go deleted file mode 100644 index 8f5eb2365..000000000 --- a/api/v1/provider.go +++ /dev/null @@ -1,37 +0,0 @@ -// Copyright © 2026 Attestant Limited. -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package v1 - -import "context" - -type providerKey struct{} - -// WithProvider returns a context with the provider address set. -// This is the universal mechanism for obtaining the provider address -// across all event types. Event structs owned by this package also -// carry the address directly in their Provider field. -func WithProvider(ctx context.Context, provider string) context.Context { - return context.WithValue(ctx, providerKey{}, provider) -} - -// Provider extracts the provider address from the context. -// It returns an empty string if no provider has been set. -func Provider(ctx context.Context) string { - provider, ok := ctx.Value(providerKey{}).(string) - if !ok { - return "" - } - - return provider -} diff --git a/api/v1/provider_test.go b/api/v1/provider_test.go deleted file mode 100644 index 7cee1df42..000000000 --- a/api/v1/provider_test.go +++ /dev/null @@ -1,62 +0,0 @@ -// Copyright © 2026 Attestant Limited. -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package v1_test - -import ( - "context" - "testing" - - v1 "github.com/attestantio/go-eth2-client/api/v1" - require "github.com/stretchr/testify/require" -) - -func TestProvider(t *testing.T) { - tests := []struct { - name string - setup func(ctx context.Context) context.Context - expected string - }{ - { - name: "Empty", - setup: func(ctx context.Context) context.Context { - return ctx - }, - expected: "", - }, - { - name: "Set", - setup: func(ctx context.Context) context.Context { - return v1.WithProvider(ctx, "http://localhost:5052") - }, - expected: "http://localhost:5052", - }, - { - name: "Override", - setup: func(ctx context.Context) context.Context { - ctx = v1.WithProvider(ctx, "http://localhost:5052") - return v1.WithProvider(ctx, "http://localhost:5053") - }, - expected: "http://localhost:5053", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - ctx := context.Background() - ctx = test.setup(ctx) - result := v1.Provider(ctx) - require.Equal(t, test.expected, result) - }) - } -} diff --git a/http/events.go b/http/events.go index a9c17e120..434a7b857 100644 --- a/http/events.go +++ b/http/events.go @@ -233,8 +233,6 @@ func (s *Service) handleAttestationEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.AttestationHandler != nil: opts.AttestationHandler(ctx, data) @@ -263,8 +261,6 @@ func (s *Service) handleAttesterSlashingEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.AttesterSlashingHandler != nil: opts.AttesterSlashingHandler(ctx, data) @@ -294,7 +290,6 @@ func (s *Service) handleBlobSidecarEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.BlobSidecarHandler != nil: @@ -325,7 +320,6 @@ func (s *Service) handleBlockEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.BlockHandler != nil: @@ -356,7 +350,6 @@ func (s *Service) handleBlockGossipEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.BlockGossipHandler != nil: @@ -386,8 +379,6 @@ func (s *Service) handleBLSToExecutionChangeEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.BLSToExecutionChangeHandler != nil: opts.BLSToExecutionChangeHandler(ctx, data) @@ -417,7 +408,6 @@ func (s *Service) handleChainReorgEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.ChainReorgHandler != nil: @@ -447,8 +437,6 @@ func (s *Service) handleContributionAndProofEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.ContributionAndProofHandler != nil: opts.ContributionAndProofHandler(ctx, data) @@ -478,7 +466,6 @@ func (s *Service) handleFinalizedCheckpointEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.FinalizedCheckpointHandler != nil: @@ -509,7 +496,6 @@ func (s *Service) handleHeadEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.HeadHandler != nil: @@ -540,7 +526,6 @@ func (s *Service) handlePayloadAttributesEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.PayloadAttributesHandler != nil: @@ -570,8 +555,6 @@ func (s *Service) handleProposerSlashingEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.ProposerSlashingHandler != nil: opts.ProposerSlashingHandler(ctx, data) @@ -600,8 +583,6 @@ func (s *Service) handleSingleAttestationEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.SingleAttestationHandler != nil: opts.SingleAttestationHandler(ctx, data) @@ -630,8 +611,6 @@ func (s *Service) handleVoluntaryExitEvent(ctx context.Context, return } - ctx = apiv1.WithProvider(ctx, s.address) - switch { case opts.VoluntaryExitHandler != nil: opts.VoluntaryExitHandler(ctx, data) @@ -661,7 +640,6 @@ func (s *Service) handleDataColumnSidecarEvent(ctx context.Context, } data.Provider = s.address - ctx = apiv1.WithProvider(ctx, s.address) switch { case opts.DataColumnSidecarHandler != nil: diff --git a/multi/events.go b/multi/events.go index 6103cb1fe..98af699ca 100644 --- a/multi/events.go +++ b/multi/events.go @@ -156,7 +156,6 @@ func (h *activeHandler) attestationHandler(ctx context.Context, data *spec.Versi log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.AttestationHandler(ctx, data) } @@ -173,7 +172,6 @@ func (h *activeHandler) attesterSlashingHandler(ctx context.Context, data *elect log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.AttesterSlashingHandler(ctx, data) } @@ -191,7 +189,6 @@ func (h *activeHandler) blobSidecarHandler(ctx context.Context, data *apiv1.Blob log.Trace().Msg("Forwarding due to primary active address") data.Provider = h.address - ctx = apiv1.WithProvider(ctx, h.address) h.opts.BlobSidecarHandler(ctx, data) } @@ -208,7 +205,6 @@ func (h *activeHandler) blsToExecutionChangeHandler(ctx context.Context, data *c log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.BLSToExecutionChangeHandler(ctx, data) } @@ -226,7 +222,6 @@ func (h *activeHandler) chainReorgHandler(ctx context.Context, data *apiv1.Chain log.Trace().Msg("Forwarding due to primary active address") data.Provider = h.address - ctx = apiv1.WithProvider(ctx, h.address) h.opts.ChainReorgHandler(ctx, data) } @@ -243,7 +238,6 @@ func (h *activeHandler) contributionAndProofHandler(ctx context.Context, data *a log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.ContributionAndProofHandler(ctx, data) } @@ -261,7 +255,6 @@ func (h *activeHandler) finalizedCheckpointHandler(ctx context.Context, data *ap log.Trace().Msg("Forwarding due to primary active address") data.Provider = h.address - ctx = apiv1.WithProvider(ctx, h.address) h.opts.FinalizedCheckpointHandler(ctx, data) } @@ -279,7 +272,6 @@ func (h *activeHandler) headHandler(ctx context.Context, data *apiv1.HeadEvent) log.Trace().Msg("Forwarding due to primary active address") data.Provider = h.address - ctx = apiv1.WithProvider(ctx, h.address) h.opts.HeadHandler(ctx, data) } @@ -297,7 +289,6 @@ func (h *activeHandler) payloadAttributesHandler(ctx context.Context, data *apiv log.Trace().Msg("Forwarding due to primary active address") data.Provider = h.address - ctx = apiv1.WithProvider(ctx, h.address) h.opts.PayloadAttributesHandler(ctx, data) } @@ -314,7 +305,6 @@ func (h *activeHandler) proposerSlashingHandler(ctx context.Context, data *phase log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.ProposerSlashingHandler(ctx, data) } @@ -331,7 +321,6 @@ func (h *activeHandler) singleAttestationHandler(ctx context.Context, data *elec log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.SingleAttestationHandler(ctx, data) } @@ -348,7 +337,6 @@ func (h *activeHandler) voluntaryExitHandler(ctx context.Context, data *phase0.S log.Trace().Msg("Forwarding due to primary active address") - ctx = apiv1.WithProvider(ctx, h.address) h.opts.VoluntaryExitHandler(ctx, data) }