Skip to content

Commit dd7e0cd

Browse files
alpealpe
committed
Linter
1 parent 04aefda commit dd7e0cd

5 files changed

Lines changed: 50 additions & 58 deletions

File tree

‎block/internal/syncing/da_follower.go‎

Lines changed: 29 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,6 @@ type daFollower struct {
6969
daBlockTime time.Duration
7070

7171
// lifecycle
72-
ctx context.Context
7372
cancel context.CancelFunc
7473
wg sync.WaitGroup
7574
}
@@ -108,11 +107,11 @@ func NewDAFollower(cfg DAFollowerConfig) DAFollower {
108107

109108
// Start begins the follow and catchup goroutines.
110109
func (f *daFollower) Start(ctx context.Context) error {
111-
f.ctx, f.cancel = context.WithCancel(ctx)
110+
ctx, f.cancel = context.WithCancel(ctx)
112111

113112
f.wg.Add(2)
114-
go f.followLoop()
115-
go f.catchupLoop()
113+
go f.followLoop(ctx)
114+
go f.catchupLoop(ctx)
116115

117116
f.logger.Info().
118117
Uint64("start_da_height", f.localDAHeight.Load()).
@@ -144,20 +143,20 @@ func (f *daFollower) signalCatchup() {
144143

145144
// followLoop subscribes to DA blob events and keeps highestSeenDAHeight up to date.
146145
// When a new height appears above localDAHeight, it wakes the catchup loop.
147-
func (f *daFollower) followLoop() {
146+
func (f *daFollower) followLoop(ctx context.Context) {
148147
defer f.wg.Done()
149148

150149
f.logger.Info().Msg("starting follow loop")
151150
defer f.logger.Info().Msg("follow loop stopped")
152151

153152
for {
154-
if err := f.runSubscription(); err != nil {
155-
if f.ctx.Err() != nil {
153+
if err := f.runSubscription(ctx); err != nil {
154+
if ctx.Err() != nil {
156155
return
157156
}
158157
f.logger.Warn().Err(err).Msg("DA subscription failed, reconnecting")
159158
select {
160-
case <-f.ctx.Done():
159+
case <-ctx.Done():
161160
return
162161
case <-time.After(f.backoff()):
163162
}
@@ -169,9 +168,9 @@ func (f *daFollower) followLoop() {
169168
// different) and processes events until a channel is closed or an error occurs.
170169
// A watchdog timer triggers if no events arrive within watchdogTimeout(),
171170
// causing a reconnect.
172-
func (f *daFollower) runSubscription() error {
171+
func (f *daFollower) runSubscription(ctx context.Context) error {
173172
// Sub-context ensures the merge goroutine is cancelled when this function returns.
174-
subCtx, subCancel := context.WithCancel(f.ctx)
173+
subCtx, subCancel := context.WithCancel(ctx)
175174
defer subCancel()
176175

177176
headerCh, err := f.client.Subscribe(subCtx, f.namespace)
@@ -201,7 +200,7 @@ func (f *daFollower) runSubscription() error {
201200
if !ok {
202201
return errors.New("subscription channel closed")
203202
}
204-
f.handleSubscriptionEvent(ev)
203+
f.handleSubscriptionEvent(ctx, ev)
205204
watchdog.Reset(watchdogTimeout)
206205
case <-watchdog.C:
207206
return errors.New("subscription watchdog: no events received, reconnecting")
@@ -251,17 +250,17 @@ func (f *daFollower) mergeSubscriptions(
251250
//
252251
// Uses CAS on localDAHeight to claim exclusive access to processBlobs,
253252
// preventing concurrent map access with catchupLoop.
254-
func (f *daFollower) handleSubscriptionEvent(ev datypes.SubscriptionEvent) {
253+
func (f *daFollower) handleSubscriptionEvent(ctx context.Context, ev datypes.SubscriptionEvent) {
255254
// Always record the highest height we've seen from the subscription.
256255
f.updateHighest(ev.Height)
257256

258257
// Fast path: try to claim this height for inline processing.
259258
// CAS(N, N+1) ensures only one goroutine (followLoop or catchupLoop)
260259
// can enter processBlobs for height N.
261260
if len(ev.Blobs) > 0 && f.localDAHeight.CompareAndSwap(ev.Height, ev.Height+1) {
262-
events := f.retriever.ProcessBlobs(f.ctx, ev.Blobs, ev.Height)
261+
events := f.retriever.ProcessBlobs(ctx, ev.Blobs, ev.Height)
263262
for _, event := range events {
264-
if err := f.pipeEvent(f.ctx, event); err != nil {
263+
if err := f.pipeEvent(ctx, event); err != nil {
265264
// Roll back so catchupLoop can retry this height.
266265
f.localDAHeight.Store(ev.Height)
267266
f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
@@ -299,27 +298,27 @@ func (f *daFollower) updateHighest(height uint64) {
299298

300299
// catchupLoop waits for signals and sequentially retrieves DA heights
301300
// from localDAHeight up to highestSeenDAHeight.
302-
func (f *daFollower) catchupLoop() {
301+
func (f *daFollower) catchupLoop(ctx context.Context) {
303302
defer f.wg.Done()
304303

305304
f.logger.Info().Msg("starting catchup loop")
306305
defer f.logger.Info().Msg("catchup loop stopped")
307306

308307
for {
309308
select {
310-
case <-f.ctx.Done():
309+
case <-ctx.Done():
311310
return
312311
case <-f.catchupSignal:
313-
f.runCatchup()
312+
f.runCatchup(ctx)
314313
}
315314
}
316315
}
317316

318317
// runCatchup sequentially retrieves from localDAHeight to highestSeenDAHeight.
319318
// It handles priority heights first, then sequential heights.
320-
func (f *daFollower) runCatchup() {
319+
func (f *daFollower) runCatchup(ctx context.Context) {
321320
for {
322-
if f.ctx.Err() != nil {
321+
if ctx.Err() != nil {
323322
return
324323
}
325324

@@ -332,8 +331,8 @@ func (f *daFollower) runCatchup() {
332331
f.logger.Debug().
333332
Uint64("da_height", priorityHeight).
334333
Msg("fetching priority DA height from P2P hint")
335-
if err := f.fetchAndPipeHeight(priorityHeight); err != nil {
336-
if !f.waitOnCatchupError(err, priorityHeight) {
334+
if err := f.fetchAndPipeHeight(ctx, priorityHeight); err != nil {
335+
if !f.waitOnCatchupError(ctx, err, priorityHeight) {
337336
return
338337
}
339338
}
@@ -350,16 +349,16 @@ func (f *daFollower) runCatchup() {
350349
return
351350
}
352351

353-
// CAS claims this height — prevents followLoop from inline-processing
352+
// CAS claims this height prevents followLoop from inline-processing
354353
if !f.localDAHeight.CompareAndSwap(local, local+1) {
355354
// followLoop already advanced past this height via inline processing.
356355
continue
357356
}
358357

359-
if err := f.fetchAndPipeHeight(local); err != nil {
358+
if err := f.fetchAndPipeHeight(ctx, local); err != nil {
360359
// Roll back so we can retry after backoff.
361360
f.localDAHeight.Store(local)
362-
if !f.waitOnCatchupError(err, local) {
361+
if !f.waitOnCatchupError(ctx, err, local) {
363362
return
364363
}
365364
continue
@@ -369,8 +368,8 @@ func (f *daFollower) runCatchup() {
369368

370369
// fetchAndPipeHeight retrieves events at a single DA height and pipes them
371370
// to the syncer.
372-
func (f *daFollower) fetchAndPipeHeight(daHeight uint64) error {
373-
events, err := f.retriever.RetrieveFromDA(f.ctx, daHeight)
371+
func (f *daFollower) fetchAndPipeHeight(ctx context.Context, daHeight uint64) error {
372+
events, err := f.retriever.RetrieveFromDA(ctx, daHeight)
374373
if err != nil {
375374
switch {
376375
case errors.Is(err, datypes.ErrBlobNotFound):
@@ -387,7 +386,7 @@ func (f *daFollower) fetchAndPipeHeight(daHeight uint64) error {
387386
}
388387

389388
for _, event := range events {
390-
if err := f.pipeEvent(f.ctx, event); err != nil {
389+
if err := f.pipeEvent(ctx, event); err != nil {
391390
return err
392391
}
393392
}
@@ -401,17 +400,17 @@ var errCaughtUp = errors.New("caught up with DA head")
401400
// waitOnCatchupError logs the error and backs off before retrying.
402401
// It returns true if the caller should continue (retry), or false if the
403402
// catchup loop should exit (context cancelled or caught-up sentinel).
404-
func (f *daFollower) waitOnCatchupError(err error, daHeight uint64) bool {
403+
func (f *daFollower) waitOnCatchupError(ctx context.Context, err error, daHeight uint64) bool {
405404
if errors.Is(err, errCaughtUp) {
406405
f.logger.Debug().Uint64("da_height", daHeight).Msg("DA catchup reached head, waiting for subscription signal")
407406
return false
408407
}
409-
if f.ctx.Err() != nil {
408+
if ctx.Err() != nil {
410409
return false
411410
}
412411
f.logger.Warn().Err(err).Uint64("da_height", daHeight).Msg("catchup error, backing off")
413412
select {
414-
case <-f.ctx.Done():
413+
case <-ctx.Done():
415414
return false
416415
case <-time.After(f.backoff()):
417416
return true

‎block/internal/syncing/syncer.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -208,7 +208,7 @@ func (s *Syncer) Start(ctx context.Context) error {
208208
StartDAHeight: s.daRetrieverHeight.Load(),
209209
DABlockTime: s.config.DA.BlockTime.Duration,
210210
})
211-
if err := s.daFollower.Start(s.ctx); err != nil {
211+
if err := s.daFollower.Start(ctx); err != nil {
212212
return fmt.Errorf("failed to start DA follower: %w", err)
213213
}
214214

‎block/internal/syncing/syncer_backoff_test.go‎

Lines changed: 10 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
7373
DABlockTime: tc.daBlockTime,
7474
}).(*daFollower)
7575

76-
follower.ctx, follower.cancel = context.WithCancel(ctx)
76+
ctx, follower.cancel = context.WithCancel(ctx)
7777
follower.highestSeenDAHeight.Store(102)
7878

7979
var callTimes []time.Time
@@ -105,7 +105,7 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
105105
Return(nil, datypes.ErrBlobNotFound).Once()
106106
}
107107

108-
go follower.runCatchup()
108+
go follower.runCatchup(ctx)
109109
<-ctx.Done()
110110

111111
if tc.expectsBackoff {
@@ -157,7 +157,7 @@ func TestDAFollower_BackoffResetOnSuccess(t *testing.T) {
157157
DABlockTime: 1 * time.Second,
158158
}).(*daFollower)
159159

160-
follower.ctx, follower.cancel = context.WithCancel(ctx)
160+
ctx, follower.cancel = context.WithCancel(ctx)
161161
follower.highestSeenDAHeight.Store(105)
162162

163163
var callTimes []time.Time
@@ -198,7 +198,7 @@ func TestDAFollower_BackoffResetOnSuccess(t *testing.T) {
198198
}).
199199
Return(nil, datypes.ErrBlobNotFound).Once()
200200

201-
go follower.runCatchup()
201+
go follower.runCatchup(ctx)
202202
<-ctx.Done()
203203

204204
require.Len(t, callTimes, 3, "should make exactly 3 calls")
@@ -234,21 +234,19 @@ func TestDAFollower_CatchupThenReachHead(t *testing.T) {
234234
DABlockTime: 500 * time.Millisecond,
235235
}).(*daFollower)
236236

237-
follower.ctx, follower.cancel = context.WithCancel(ctx)
238237
follower.highestSeenDAHeight.Store(5)
239238

240239
var fetchedHeights []uint64
241240

242241
for h := uint64(3); h <= 5; h++ {
243-
h := h
244242
daRetriever.On("RetrieveFromDA", mock.Anything, h).
245243
Run(func(args mock.Arguments) {
246244
fetchedHeights = append(fetchedHeights, h)
247245
}).
248246
Return(nil, datypes.ErrBlobNotFound).Once()
249247
}
250248

251-
follower.runCatchup()
249+
follower.runCatchup(ctx)
252250

253251
assert.True(t, follower.HasReachedHead(), "should have reached DA head")
254252
// Heights 3, 4, 5 processed; local now at 6 which > highest (5) → caught up
@@ -278,9 +276,6 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
278276
DABlockTime: 500 * time.Millisecond,
279277
}).(*daFollower)
280278

281-
follower.ctx, follower.cancel = context.WithCancel(t.Context())
282-
defer follower.cancel()
283-
284279
blobs := [][]byte{[]byte("header-blob"), []byte("data-blob")}
285280
expectedEvents := []common.DAHeightEvent{
286281
{DaHeight: 10, Source: common.SourceDA},
@@ -291,7 +286,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
291286
Return(expectedEvents).Once()
292287

293288
// Simulate subscription event at the current localDAHeight
294-
follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
289+
follower.handleSubscriptionEvent(t.Context(), datypes.SubscriptionEvent{
295290
Height: 10,
296291
Blobs: blobs,
297292
})
@@ -317,11 +312,12 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
317312
DABlockTime: 500 * time.Millisecond,
318313
}).(*daFollower)
319314

320-
follower.ctx, follower.cancel = context.WithCancel(t.Context())
315+
ctx := t.Context()
316+
ctx, follower.cancel = context.WithCancel(ctx)
321317
defer follower.cancel()
322318

323319
// Subscription reports height 15 but local is at 10 — should NOT process inline
324-
follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
320+
follower.handleSubscriptionEvent(ctx, datypes.SubscriptionEvent{
325321
Height: 15,
326322
Blobs: [][]byte{[]byte("blob")},
327323
})
@@ -346,11 +342,8 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
346342
DABlockTime: 500 * time.Millisecond,
347343
}).(*daFollower)
348344

349-
follower.ctx, follower.cancel = context.WithCancel(t.Context())
350-
defer follower.cancel()
351-
352345
// Subscription at current height but no blobs — should fall through
353-
follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
346+
follower.handleSubscriptionEvent(t.Context(), datypes.SubscriptionEvent{
354347
Height: 10,
355348
Blobs: nil,
356349
})

‎block/internal/syncing/syncer_benchmark_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,8 @@ func BenchmarkSyncerIO(b *testing.B) {
4343
fixt := newBenchFixture(b, spec.heights, spec.shuffledTx, spec.daDelay, spec.execDelay, true)
4444

4545
// run both loops
46-
go fixt.s.processLoop(b.Context())
46+
ctx := b.Context()
47+
go fixt.s.processLoop(ctx)
4748

4849
// Create a DAFollower to drive DA retrieval.
4950
follower := NewDAFollower(DAFollowerConfig{
@@ -54,14 +55,13 @@ func BenchmarkSyncerIO(b *testing.B) {
5455
StartDAHeight: fixt.s.daRetrieverHeight.Load(),
5556
DABlockTime: 0,
5657
}).(*daFollower)
57-
follower.ctx, follower.cancel = context.WithCancel(fixt.s.ctx)
5858
follower.highestSeenDAHeight.Store(spec.heights + daHeightOffset)
59-
go follower.runCatchup()
59+
go follower.runCatchup(ctx)
6060

61-
fixt.s.startSyncWorkers(b.Context())
61+
fixt.s.startSyncWorkers(ctx)
6262

6363
require.Eventually(b, func() bool {
64-
processedHeight, _ := fixt.s.store.Height(b.Context())
64+
processedHeight, _ := fixt.s.store.Height(ctx)
6565
return processedHeight == spec.heights
6666
}, 5*time.Second, 50*time.Microsecond)
6767
fixt.s.cancel()
@@ -75,7 +75,7 @@ func BenchmarkSyncerIO(b *testing.B) {
7575
require.Len(b, fixt.s.heightInCh, 0)
7676

7777
assert.Equal(b, spec.heights+daHeightOffset, fixt.s.daRetrieverHeight)
78-
gotStoreHeight, err := fixt.s.store.Height(b.Context())
78+
gotStoreHeight, err := fixt.s.store.Height(ctx)
7979
require.NoError(b, err)
8080
assert.Equal(b, spec.heights, gotStoreHeight)
8181
}

‎block/internal/syncing/syncer_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -427,10 +427,10 @@ func TestSyncLoopPersistState(t *testing.T) {
427427
StartDAHeight: syncerInst1.daRetrieverHeight.Load(),
428428
DABlockTime: cfg.DA.BlockTime.Duration,
429429
}).(*daFollower)
430-
follower1.ctx, follower1.cancel = context.WithCancel(ctx)
430+
ctx, follower1.cancel = context.WithCancel(ctx)
431431
// Set highest so catchup runs through all mocked heights.
432432
follower1.highestSeenDAHeight.Store(myFutureDAHeight)
433-
go follower1.runCatchup()
433+
go follower1.runCatchup(ctx)
434434
syncerInst1.startSyncWorkers(ctx)
435435
syncerInst1.wg.Wait()
436436
requireEmptyChan(t, errorCh)
@@ -504,9 +504,9 @@ func TestSyncLoopPersistState(t *testing.T) {
504504
StartDAHeight: syncerInst2.daRetrieverHeight.Load(),
505505
DABlockTime: cfg.DA.BlockTime.Duration,
506506
}).(*daFollower)
507-
follower2.ctx, follower2.cancel = context.WithCancel(ctx)
507+
ctx, follower2.cancel = context.WithCancel(ctx)
508508
follower2.highestSeenDAHeight.Store(syncerInst2.daRetrieverHeight.Load() + 1)
509-
go follower2.runCatchup()
509+
go follower2.runCatchup(ctx)
510510
syncerInst2.startSyncWorkers(ctx)
511511
syncerInst2.wg.Wait()
512512

0 commit comments

Comments
 (0)