Skip to content

Commit 341a37b

Browse files
julienrbrtjulienrbrt
committed
clone nicely
1 parent 1f75ad7 commit 341a37b

7 files changed

Lines changed: 50 additions & 36 deletions

File tree

‎block/aggregation.go‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import (
77
)
88

99
// AggregationLoop is responsible for aggregating transactions into rollup-blocks.
10-
func (m *Manager) AggregationLoop(ctx context.Context) {
10+
func (m *Manager) AggregationLoop(ctx context.Context, errCh chan<- error) {
1111
initialHeight := m.genesis.InitialHeight //nolint:gosec
1212
height, err := m.store.Height(ctx)
1313
if err != nil {
@@ -39,14 +39,13 @@ func (m *Manager) AggregationLoop(ctx context.Context) {
3939
// transactions or every LazyBlockTime.
4040
if m.config.Node.LazyMode {
4141
if err := m.lazyAggregationLoop(ctx, blockTimer); err != nil {
42-
panic(fmt.Errorf("error in lazy aggregation loop: %w", err))
42+
errCh <- fmt.Errorf("error in lazy aggregation loop: %w", err)
4343
}
44-
4544
return
4645
}
4746

4847
if err := m.normalAggregationLoop(ctx, blockTimer); err != nil {
49-
panic(fmt.Errorf("error in normal aggregation loop: %w", err))
48+
errCh <- fmt.Errorf("error in normal aggregation loop: %w", err)
5049
}
5150
}
5251

‎block/aggregation_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ func TestAggregationLoop_Normal_BasicInterval(t *testing.T) {
8282
wg.Add(1)
8383
go func() {
8484
defer wg.Done()
85-
m.AggregationLoop(ctx)
85+
m.AggregationLoop(ctx, make(chan<- error))
8686
m.logger.Info("AggregationLoop exited")
8787
}()
8888

@@ -188,7 +188,7 @@ func TestAggregationLoop_Normal_PublishBlockError(t *testing.T) {
188188
wg.Add(1)
189189
go func() {
190190
defer wg.Done()
191-
m.AggregationLoop(ctx)
191+
m.AggregationLoop(ctx, make(chan<- error))
192192
m.logger.Info("AggregationLoop exited")
193193
}()
194194

‎block/da_includer.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import (
88

99
// DAIncluderLoop is responsible for advancing the DAIncludedHeight by checking if blocks after the current height
1010
// have both their header and data marked as DA-included in the caches. If so, it calls setDAIncludedHeight.
11-
func (m *Manager) DAIncluderLoop(ctx context.Context) {
11+
func (m *Manager) DAIncluderLoop(ctx context.Context, errCh chan<- error) {
1212
for {
1313
select {
1414
case <-ctx.Done():
@@ -28,8 +28,10 @@ func (m *Manager) DAIncluderLoop(ctx context.Context) {
2828
if daIncluded {
2929
// Both header and data are DA-included, so we can advance the height
3030
if err := m.incrementDAIncludedHeight(ctx); err != nil {
31-
panic(fmt.Errorf("error while incrementing DA included height: %w", err))
31+
errCh <- fmt.Errorf("error while incrementing DA included height: %w", err)
32+
return
3233
}
34+
3335
currentDAIncluded = nextHeight
3436
} else {
3537
// Stop at the first block that is not DA-included

‎block/da_includer_test.go‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ func TestDAIncluderLoop_AdvancesHeightWhenBothDAIncluded(t *testing.T) {
6767

6868
go func() {
6969
defer wg.Done()
70-
m.DAIncluderLoop(ctx)
70+
m.DAIncluderLoop(ctx, make(chan<- error))
7171
}()
7272

7373
m.sendNonBlockingSignalToDAIncluderCh()
@@ -100,7 +100,7 @@ func TestDAIncluderLoop_StopsWhenHeaderNotDAIncluded(t *testing.T) {
100100
wg.Add(1)
101101
go func() {
102102
defer wg.Done()
103-
m.DAIncluderLoop(ctx)
103+
m.DAIncluderLoop(ctx, make(chan<- error))
104104
}()
105105

106106
m.sendNonBlockingSignalToDAIncluderCh()
@@ -132,7 +132,7 @@ func TestDAIncluderLoop_StopsWhenDataNotDAIncluded(t *testing.T) {
132132
wg.Add(1)
133133
go func() {
134134
defer wg.Done()
135-
m.DAIncluderLoop(ctx)
135+
m.DAIncluderLoop(ctx, make(chan<- error))
136136
}()
137137

138138
m.sendNonBlockingSignalToDAIncluderCh()
@@ -163,7 +163,7 @@ func TestDAIncluderLoop_StopsOnGetBlockDataError(t *testing.T) {
163163
wg.Add(1)
164164
go func() {
165165
defer wg.Done()
166-
m.DAIncluderLoop(ctx)
166+
m.DAIncluderLoop(ctx, make(chan<- error))
167167
}()
168168

169169
m.sendNonBlockingSignalToDAIncluderCh()
@@ -286,7 +286,7 @@ func TestDAIncluderLoop_MultipleConsecutiveHeightsDAIncluded(t *testing.T) {
286286
wg.Add(1)
287287
go func() {
288288
defer wg.Done()
289-
m.DAIncluderLoop(ctx)
289+
m.DAIncluderLoop(ctx, make(chan<- error))
290290
}()
291291

292292
m.sendNonBlockingSignalToDAIncluderCh()
@@ -325,7 +325,7 @@ func TestDAIncluderLoop_AdvancesHeightWhenDataHashIsEmptyAndHeaderDAIncluded(t *
325325
wg.Add(1)
326326
go func() {
327327
defer wg.Done()
328-
m.DAIncluderLoop(ctx)
328+
m.DAIncluderLoop(ctx, make(chan<- error))
329329
}()
330330

331331
m.sendNonBlockingSignalToDAIncluderCh()
@@ -357,7 +357,7 @@ func TestDAIncluderLoop_DoesNotAdvanceWhenDataHashIsEmptyAndHeaderNotDAIncluded(
357357
wg.Add(1)
358358
go func() {
359359
defer wg.Done()
360-
m.DAIncluderLoop(ctx)
360+
m.DAIncluderLoop(ctx, make(chan<- error))
361361
}()
362362

363363
m.sendNonBlockingSignalToDAIncluderCh()

‎block/sync.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import (
1212
// SyncLoop is responsible for syncing blocks.
1313
//
1414
// SyncLoop processes headers gossiped in P2P network to know what's the latest block height, block data is retrieved from DA layer.
15-
func (m *Manager) SyncLoop(ctx context.Context) {
15+
func (m *Manager) SyncLoop(ctx context.Context, errCh chan<- error) {
1616
daTicker := time.NewTicker(m.config.DA.BlockTime.Duration)
1717
defer daTicker.Stop()
1818
blockTicker := time.NewTicker(m.config.Node.BlockTime.Duration)
@@ -63,7 +63,8 @@ func (m *Manager) SyncLoop(ctx context.Context) {
6363
}
6464

6565
if err = m.trySyncNextBlock(ctx, daHeight); err != nil {
66-
panic(fmt.Errorf("failed to sync next block: %w", err))
66+
errCh <- fmt.Errorf("failed to sync next block: %w", err)
67+
return
6768
}
6869

6970
m.headerCache.SetSeen(headerHash)
@@ -113,7 +114,8 @@ func (m *Manager) SyncLoop(ctx context.Context) {
113114

114115
err = m.trySyncNextBlock(ctx, daHeight)
115116
if err != nil {
116-
panic(fmt.Errorf("failed to sync next block: %w", err))
117+
errCh <- fmt.Errorf("failed to sync next block: %w", err)
118+
return
117119
}
118120
m.dataCache.SetSeen(dataHash)
119121
case <-ctx.Done():

‎block/sync_test.go‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,7 @@ func TestSyncLoop_ProcessSingleBlock_HeaderFirst(t *testing.T) {
139139
wg.Add(1)
140140
go func() {
141141
defer wg.Done()
142-
m.SyncLoop(ctx)
142+
m.SyncLoop(ctx, make(chan<- error))
143143
}()
144144

145145
t.Logf("Sending header event for height %d", newHeight)
@@ -226,7 +226,7 @@ func TestSyncLoop_ProcessSingleBlock_DataFirst(t *testing.T) {
226226
wg.Add(1)
227227
go func() {
228228
defer wg.Done()
229-
m.SyncLoop(ctx)
229+
m.SyncLoop(ctx, make(chan<- error))
230230
}()
231231

232232
t.Logf("Sending data event for height %d", newHeight)
@@ -351,7 +351,7 @@ func TestSyncLoop_ProcessMultipleBlocks_Sequentially(t *testing.T) {
351351
wg.Add(1)
352352
go func() {
353353
defer wg.Done()
354-
m.SyncLoop(ctx)
354+
m.SyncLoop(ctx, make(chan<- error))
355355
t.Log("SyncLoop exited.")
356356
}()
357357

@@ -498,7 +498,7 @@ func TestSyncLoop_ProcessBlocks_OutOfOrderArrival(t *testing.T) {
498498
wg.Add(1)
499499
go func() {
500500
defer wg.Done()
501-
m.SyncLoop(ctx)
501+
m.SyncLoop(ctx, make(chan<- error))
502502
t.Log("SyncLoop exited.")
503503
}()
504504

@@ -605,7 +605,7 @@ func TestSyncLoop_IgnoreDuplicateEvents(t *testing.T) {
605605
wg.Add(1)
606606
go func() {
607607
defer wg.Done()
608-
m.SyncLoop(ctx)
608+
m.SyncLoop(ctx, make(chan<- error))
609609
t.Log("SyncLoop exited.")
610610
}()
611611

@@ -703,7 +703,7 @@ func TestSyncLoop_PanicOnApplyError(t *testing.T) {
703703
t.Error("SyncLoop did not panic as expected")
704704
}
705705
}()
706-
m.SyncLoop(ctx)
706+
m.SyncLoop(ctx, make(chan<- error))
707707
t.Log("SyncLoop exited.")
708708
}()
709709

@@ -821,7 +821,7 @@ func TestDataCommitmentToHeight_HeaderThenData(t *testing.T) {
821821
wg.Add(1)
822822
go func() {
823823
defer wg.Done()
824-
m.SyncLoop(ctx)
824+
m.SyncLoop(ctx, make(chan<- error))
825825
}()
826826

827827
headerInCh <- NewHeaderEvent{Header: header, DAHeight: initialState.DAHeight}
@@ -868,7 +868,7 @@ func TestDataCommitmentToHeight_HeaderWithExistingData(t *testing.T) {
868868
wg.Add(1)
869869
go func() {
870870
defer wg.Done()
871-
m.SyncLoop(ctx)
871+
m.SyncLoop(ctx, make(chan<- error))
872872
}()
873873

874874
headerInCh <- NewHeaderEvent{Header: header, DAHeight: initialState.DAHeight}
@@ -911,7 +911,7 @@ func TestDataCommitmentToHeight_DataBeforeHeader(t *testing.T) {
911911
wg.Add(1)
912912
go func() {
913913
defer wg.Done()
914-
m.SyncLoop(ctx)
914+
m.SyncLoop(ctx, make(chan<- error))
915915
}()
916916

917917
data.Metadata = nil
@@ -961,7 +961,7 @@ func TestDataCommitmentToHeight_HeaderAlreadySynced(t *testing.T) {
961961
wg.Add(1)
962962
go func() {
963963
defer wg.Done()
964-
m.SyncLoop(ctx)
964+
m.SyncLoop(ctx, make(chan<- error))
965965
}()
966966

967967
headerInCh <- NewHeaderEvent{Header: header, DAHeight: initialState.DAHeight}
@@ -1007,7 +1007,7 @@ func TestDataCommitmentToHeight_EmptyDataHash(t *testing.T) {
10071007
wg.Add(1)
10081008
go func() {
10091009
defer wg.Done()
1010-
m.SyncLoop(ctx)
1010+
m.SyncLoop(ctx, make(chan<- error))
10111011
}()
10121012

10131013
headerInCh <- NewHeaderEvent{Header: header, DAHeight: initialState.DAHeight}
@@ -1043,7 +1043,7 @@ func TestDataCommitmentToHeight_DataAlreadySeen(t *testing.T) {
10431043
wg.Add(1)
10441044
go func() {
10451045
defer wg.Done()
1046-
m.SyncLoop(ctx)
1046+
m.SyncLoop(ctx, make(chan<- error))
10471047
}()
10481048

10491049
dataInCh <- NewDataEvent{Data: data, DAHeight: initialState.DAHeight}

‎node/full.go‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -395,26 +395,37 @@ func (n *FullNode) Run(ctx context.Context) error {
395395
return fmt.Errorf("error while starting data sync service: %w", err)
396396
}
397397

398+
errCh := make(chan error, 1)
399+
398400
if n.nodeConfig.Node.Aggregator {
399401
n.Logger.Info("working in aggregator mode", "block time", n.nodeConfig.Node.BlockTime)
400-
go n.blockManager.AggregationLoop(ctx)
402+
go n.blockManager.AggregationLoop(ctx, errCh)
401403
go n.reaper.Start(ctx)
402404
go n.blockManager.HeaderSubmissionLoop(ctx)
403405
go n.blockManager.BatchSubmissionLoop(ctx)
404406
go n.headerPublishLoop(ctx)
405407
go n.dataPublishLoop(ctx)
406-
go n.blockManager.DAIncluderLoop(ctx)
408+
go n.blockManager.DAIncluderLoop(ctx, errCh)
407409
} else {
408410
go n.blockManager.RetrieveLoop(ctx)
409411
go n.blockManager.HeaderStoreRetrieveLoop(ctx)
410412
go n.blockManager.DataStoreRetrieveLoop(ctx)
411-
go n.blockManager.SyncLoop(ctx)
412-
go n.blockManager.DAIncluderLoop(ctx)
413+
go n.blockManager.SyncLoop(ctx, errCh)
414+
go n.blockManager.DAIncluderLoop(ctx, errCh)
413415
}
414416

415-
// Block until context is canceled
416-
<-ctx.Done()
417+
select {
418+
case err := <-errCh:
419+
if err != nil {
420+
n.Logger.Error("unrecoverable error in one of the go routines...", "error", err)
421+
goto Cleanup
422+
}
423+
case <-ctx.Done():
424+
// Block until context is canceled
425+
n.Logger.Info("context canceled, stopping node")
426+
}
417427

428+
Cleanup:
418429
// Perform cleanup
419430
n.Logger.Info("halting full node...")
420431
n.Logger.Info("shutting down full node sub services...")

0 commit comments

Comments
 (0)