Skip to content

Commit e918607

Browse files
alpealpe
committed
Updates
1 parent d3bc9ed commit e918607

7 files changed

Lines changed: 300 additions & 423 deletions

File tree

‎apps/evm/server/force_inclusion_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ func (m *mockDA) Get(ctx context.Context, ids []da.ID, namespace []byte) ([]da.B
5050
return nil, nil
5151
}
5252

53-
func (m *mockDA) Subscribe(_ context.Context, _ []byte) (<-chan da.SubscriptionEvent, error) {
53+
func (m *mockDA) Subscribe(_ context.Context, _ []byte, _ bool) (<-chan da.SubscriptionEvent, error) {
5454
// Not needed in these tests; return a closed channel.
5555
ch := make(chan da.SubscriptionEvent)
5656
close(ch)

‎block/internal/da/async_block_retriever.go‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,9 @@ func (f *asyncBlockRetriever) HandleEvent(ctx context.Context, ev datypes.Subscr
185185
// Also applies the prefetch window for speculative forward fetching.
186186
func (f *asyncBlockRetriever) HandleCatchup(ctx context.Context, daHeight uint64) error {
187187
if _, err := f.cache.Get(ctx, newBlockDataKey(daHeight)); err != nil {
188-
f.fetchAndCacheBlock(ctx, daHeight)
188+
if err := f.fetchAndCacheBlock(ctx, daHeight); err != nil {
189+
return err
190+
}
189191
}
190192
// Speculatively prefetch ahead.
191193
target := daHeight + f.prefetchWindow
@@ -196,22 +198,24 @@ func (f *asyncBlockRetriever) HandleCatchup(ctx context.Context, daHeight uint64
196198
if _, err := f.cache.Get(ctx, newBlockDataKey(h)); err == nil {
197199
continue // Already cached.
198200
}
199-
f.fetchAndCacheBlock(ctx, h)
201+
if err := f.fetchAndCacheBlock(ctx, h); err != nil {
202+
return err
203+
}
200204
}
201205

202206
return nil
203207
}
204208

205209
// fetchAndCacheBlock fetches a block via Retrieve and caches it.
206-
func (f *asyncBlockRetriever) fetchAndCacheBlock(ctx context.Context, height uint64) {
210+
func (f *asyncBlockRetriever) fetchAndCacheBlock(ctx context.Context, height uint64) error {
207211
f.logger.Debug().Uint64("height", height).Msg("prefetching block")
208212

209213
result := f.client.Retrieve(ctx, height, f.namespace)
210214

211215
switch result.Code {
212216
case datypes.StatusHeightFromFuture:
213217
f.logger.Debug().Uint64("height", height).Msg("block height not yet available - will retry")
214-
return
218+
return datypes.ErrHeightFromFuture
215219
case datypes.StatusNotFound:
216220
f.cacheBlock(ctx, height, result.Timestamp, nil)
217221
case datypes.StatusSuccess:
@@ -228,6 +232,7 @@ func (f *asyncBlockRetriever) fetchAndCacheBlock(ctx context.Context, height uin
228232
Str("status", result.Message).
229233
Msg("failed to retrieve block - will retry")
230234
}
235+
return nil
231236
}
232237

233238
// cacheBlock serializes and stores a block in the in-memory cache.

‎block/internal/da/async_block_retriever_test.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -177,7 +177,7 @@ func TestAsyncBlockRetriever_HeightFromFuture(t *testing.T) {
177177
// Wait a bit for catchup to attempt fetches.
178178
require.Eventually(t, func() bool {
179179
return fetcher.(*asyncBlockRetriever).subscriber.HasReachedHead()
180-
}, 1250*time.Millisecond, time.Millisecond)
180+
}, 1250*time.Second, time.Millisecond)
181181

182182
// Cache should be empty since all heights are from the future.
183183
block, err := fetcher.GetCachedBlock(ctx, 100)
@@ -191,6 +191,9 @@ func TestAsyncBlockRetriever_StopGracefully(t *testing.T) {
191191

192192
blockCh := make(chan datypes.SubscriptionEvent)
193193
client.On("Subscribe", mock.Anything, fiNs, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(blockCh), nil).Maybe()
194+
client.On("Retrieve", mock.Anything, mock.Anything, fiNs).Return(datypes.ResultRetrieve{
195+
BaseResult: datypes.BaseResult{Code: datypes.StatusHeightFromFuture},
196+
}).Maybe()
194197

195198
logger := zerolog.Nop()
196199
fetcher := NewAsyncBlockRetriever(client, logger, fiNs, 100*time.Millisecond, 100, 10)

‎block/internal/da/subscriber.go‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ func NewSubscriber(cfg SubscriberConfig) *Subscriber {
9292
fetchBlockTimestamp: cfg.FetchBlockTimestamp,
9393
}
9494
s.localDAHeight.Store(cfg.StartHeight)
95+
s.highestSeenDAHeight.Store(cfg.StartHeight)
9596
s.catchupSignal <- struct{}{}
9697

9798
if len(s.namespaces) == 0 {
@@ -319,11 +320,6 @@ func (s *Subscriber) runCatchup(ctx context.Context) {
319320
}
320321

321322
local := s.localDAHeight.Load()
322-
if local > s.highestSeenDAHeight.Load() {
323-
s.headReached.Store(true)
324-
return
325-
}
326-
327323
// CAS claims this height — prevents followLoop from inline-processing.
328324
if !s.localDAHeight.CompareAndSwap(local, local+1) {
329325
// followLoop already advanced past this height via inline processing.

‎block/internal/da/subscriber_test.go‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
package da
2+
3+
import (
4+
"context"
5+
"errors"
6+
"testing"
7+
"time"
8+
9+
"github.com/rs/zerolog"
10+
"github.com/stretchr/testify/assert"
11+
"github.com/stretchr/testify/mock"
12+
13+
datypes "github.com/evstack/ev-node/pkg/da/types"
14+
testmocks "github.com/evstack/ev-node/test/mocks"
15+
)
16+
17+
// MockSubscriberHandler mocks SubscriberHandler
18+
type MockSubscriberHandler struct {
19+
mock.Mock
20+
}
21+
22+
func (m *MockSubscriberHandler) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error {
23+
args := m.Called(ctx, ev, isInline)
24+
return args.Error(0)
25+
}
26+
27+
func (m *MockSubscriberHandler) HandleCatchup(ctx context.Context, height uint64) error {
28+
args := m.Called(ctx, height)
29+
return args.Error(0)
30+
}
31+
32+
func TestSubscriber_RunCatchup(t *testing.T) {
33+
t.Run("success_sequence", func(t *testing.T) {
34+
ctx, cancel := context.WithCancel(t.Context())
35+
defer cancel()
36+
37+
mockHandler := new(MockSubscriberHandler)
38+
mockClient := testmocks.NewMockClient(t)
39+
40+
sub := NewSubscriber(SubscriberConfig{
41+
Client: mockClient,
42+
Logger: zerolog.Nop(),
43+
Handler: mockHandler,
44+
Namespaces: [][]byte{[]byte("ns")},
45+
StartHeight: 100,
46+
DABlockTime: time.Millisecond,
47+
})
48+
49+
// It should try 100, 101, then we return ErrHeightFromFuture at 102
50+
mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).Return(nil).Once()
51+
mockHandler.On("HandleCatchup", mock.Anything, uint64(101)).Return(nil).Once()
52+
mockHandler.On("HandleCatchup", mock.Anything, uint64(102)).Return(datypes.ErrHeightFromFuture).Once()
53+
54+
sub.runCatchup(ctx)
55+
56+
mockHandler.AssertExpectations(t)
57+
assert.Equal(t, uint64(102), sub.LocalDAHeight())
58+
assert.True(t, sub.HasReachedHead())
59+
})
60+
61+
t.Run("backoff_on_error", func(t *testing.T) {
62+
ctx, cancel := context.WithCancel(t.Context())
63+
defer cancel()
64+
65+
mockHandler := new(MockSubscriberHandler)
66+
mockClient := testmocks.NewMockClient(t)
67+
68+
sub := NewSubscriber(SubscriberConfig{
69+
Client: mockClient,
70+
Logger: zerolog.Nop(),
71+
Handler: mockHandler,
72+
Namespaces: [][]byte{[]byte("ns")},
73+
StartHeight: 100,
74+
DABlockTime: time.Millisecond,
75+
})
76+
77+
var callCount int
78+
79+
mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).
80+
Run(func(args mock.Arguments) {
81+
callCount++
82+
}).
83+
Return(errors.New("network failure")).Once()
84+
85+
mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).
86+
Run(func(args mock.Arguments) {
87+
callCount++
88+
cancel()
89+
}).
90+
Return(datypes.ErrHeightFromFuture).Once()
91+
92+
sub.runCatchup(ctx)
93+
94+
mockHandler.AssertExpectations(t)
95+
assert.Equal(t, 2, callCount)
96+
assert.Equal(t, uint64(100), sub.LocalDAHeight(), "should roll back to 100 on future error")
97+
})
98+
}

‎block/internal/syncing/da_follower_test.go‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,187 @@
1+
package syncing
2+
3+
import (
4+
"context"
5+
"errors"
6+
"testing"
7+
8+
"github.com/rs/zerolog"
9+
"github.com/stretchr/testify/assert"
10+
"github.com/stretchr/testify/mock"
11+
"github.com/stretchr/testify/require"
12+
13+
"github.com/evstack/ev-node/block/internal/common"
14+
datypes "github.com/evstack/ev-node/pkg/da/types"
15+
)
16+
17+
func TestDAFollower_HandleEvent(t *testing.T) {
18+
tests := []struct {
19+
name string
20+
isInline bool
21+
blobs [][]byte
22+
mockEvents []common.DAHeightEvent
23+
mockPipeErr error
24+
expectedError string
25+
}{
26+
{
27+
name: "ignore_not_inline",
28+
isInline: false,
29+
},
30+
{
31+
name: "error_no_blobs",
32+
isInline: true,
33+
blobs: [][]byte{},
34+
expectedError: "skip inline: no blobs",
35+
},
36+
{
37+
name: "error_no_complete_events",
38+
isInline: true,
39+
blobs: [][]byte{[]byte("blob")},
40+
mockEvents: []common.DAHeightEvent{},
41+
expectedError: "skip inline: no complete events",
42+
},
43+
{
44+
name: "error_pipe_fails",
45+
isInline: true,
46+
blobs: [][]byte{[]byte("blob")},
47+
mockEvents: []common.DAHeightEvent{{DaHeight: 100}},
48+
mockPipeErr: errors.New("pipe error"),
49+
expectedError: "pipe error",
50+
},
51+
{
52+
name: "success",
53+
isInline: true,
54+
blobs: [][]byte{[]byte("blob")},
55+
mockEvents: []common.DAHeightEvent{{DaHeight: 100}},
56+
mockPipeErr: nil,
57+
},
58+
}
59+
60+
for _, tt := range tests {
61+
t.Run(tt.name, func(t *testing.T) {
62+
daRetriever := NewMockDARetriever(t)
63+
ctx := t.Context()
64+
65+
var pipedEvents []common.DAHeightEvent
66+
pipeEvent := func(_ context.Context, ev common.DAHeightEvent) error {
67+
pipedEvents = append(pipedEvents, ev)
68+
return tt.mockPipeErr
69+
}
70+
71+
follower := &daFollower{
72+
retriever: daRetriever,
73+
eventSink: common.EventSinkFunc(pipeEvent),
74+
logger: zerolog.Nop(),
75+
}
76+
77+
ev := datypes.SubscriptionEvent{Height: 100, Blobs: tt.blobs}
78+
79+
if tt.isInline && len(tt.blobs) > 0 {
80+
daRetriever.On("ProcessBlobs", mock.Anything, tt.blobs, uint64(100)).Return(tt.mockEvents)
81+
}
82+
83+
err := follower.HandleEvent(ctx, ev, tt.isInline)
84+
85+
if tt.expectedError != "" {
86+
require.Error(t, err)
87+
assert.Contains(t, err.Error(), tt.expectedError)
88+
} else {
89+
require.NoError(t, err)
90+
if tt.isInline && len(tt.blobs) > 0 {
91+
assert.Len(t, pipedEvents, len(tt.mockEvents))
92+
}
93+
}
94+
})
95+
}
96+
}
97+
98+
func TestDAFollower_HandleCatchup(t *testing.T) {
99+
t.Run("normal_sequential_fetch_success", func(t *testing.T) {
100+
daRetriever := NewMockDARetriever(t)
101+
ctx := t.Context()
102+
103+
var pipedEvents []common.DAHeightEvent
104+
pipeEvent := func(_ context.Context, ev common.DAHeightEvent) error {
105+
pipedEvents = append(pipedEvents, ev)
106+
return nil
107+
}
108+
109+
follower := &daFollower{
110+
retriever: daRetriever,
111+
eventSink: common.EventSinkFunc(pipeEvent),
112+
logger: zerolog.Nop(),
113+
priorityHeights: make([]uint64, 0),
114+
}
115+
116+
events := []common.DAHeightEvent{{DaHeight: 100}}
117+
daRetriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(events, nil)
118+
119+
err := follower.HandleCatchup(ctx, 100)
120+
require.NoError(t, err)
121+
assert.Len(t, pipedEvents, 1)
122+
})
123+
124+
t.Run("normal_sequential_fetch_blob_not_found_ignored", func(t *testing.T) {
125+
daRetriever := NewMockDARetriever(t)
126+
ctx := t.Context()
127+
128+
follower := &daFollower{
129+
retriever: daRetriever,
130+
eventSink: common.EventSinkFunc(func(_ context.Context, _ common.DAHeightEvent) error { return nil }),
131+
logger: zerolog.Nop(),
132+
priorityHeights: make([]uint64, 0),
133+
}
134+
135+
daRetriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(nil, datypes.ErrBlobNotFound)
136+
137+
err := follower.HandleCatchup(ctx, 100)
138+
require.NoError(t, err)
139+
})
140+
141+
t.Run("normal_sequential_fetch_error_propagated", func(t *testing.T) {
142+
daRetriever := NewMockDARetriever(t)
143+
ctx := t.Context()
144+
145+
follower := &daFollower{
146+
retriever: daRetriever,
147+
eventSink: common.EventSinkFunc(func(_ context.Context, _ common.DAHeightEvent) error { return nil }),
148+
logger: zerolog.Nop(),
149+
priorityHeights: make([]uint64, 0),
150+
}
151+
152+
daRetriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return(nil, datypes.ErrHeightFromFuture)
153+
154+
err := follower.HandleCatchup(ctx, 100)
155+
require.ErrorIs(t, err, datypes.ErrHeightFromFuture)
156+
})
157+
158+
t.Run("priority_queue_handled_first", func(t *testing.T) {
159+
daRetriever := NewMockDARetriever(t)
160+
ctx := t.Context()
161+
162+
var pipedEvents []common.DAHeightEvent
163+
pipeEvent := func(_ context.Context, ev common.DAHeightEvent) error {
164+
pipedEvents = append(pipedEvents, ev)
165+
return nil
166+
}
167+
168+
follower := &daFollower{
169+
retriever: daRetriever,
170+
eventSink: common.EventSinkFunc(pipeEvent),
171+
logger: zerolog.Nop(),
172+
priorityHeights: []uint64{105}, // Priority height to fetch
173+
}
174+
175+
daRetriever.On("RetrieveFromDA", mock.Anything, uint64(105)).Return([]common.DAHeightEvent{{DaHeight: 105}}, nil).Once()
176+
daRetriever.On("RetrieveFromDA", mock.Anything, uint64(100)).Return([]common.DAHeightEvent{{DaHeight: 100}}, nil).Once()
177+
178+
err := follower.HandleCatchup(ctx, 100)
179+
require.NoError(t, err)
180+
181+
// It should pipe 105 then 100
182+
assert.Len(t, pipedEvents, 2)
183+
assert.Equal(t, uint64(105), pipedEvents[0].DaHeight)
184+
assert.Equal(t, uint64(100), pipedEvents[1].DaHeight)
185+
assert.Empty(t, follower.priorityHeights)
186+
})
187+
}

0 commit comments

Comments
 (0)