Skip to content

refactor: halt at app failure#2268

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
julienrbrt merged 9 commits into from
May 20, 2025
Merged

refactor: halt at app failure #2268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
5e77da5
refactor: halt at app failure
julienrbrt May 15, 2025
d4d6fc3
updates
julienrbrt May 15, 2025
1f75ad7
lint + update
julienrbrt May 15, 2025
341a37b
clone nicely
julienrbrt May 16, 2025
c6a785f
feedback
julienrbrt May 19, 2025
900fb06
Merge branch 'main' into julien/halt
julienrbrt May 19, 2025
e5dcf6b
fix merge
julienrbrt May 19, 2025
3db6aca
updates
julienrbrt May 20, 2025
5cdd35f
Merge branch 'main' into julien/halt
julienrbrt May 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 27 additions & 15 deletions block/aggregation.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2,11 +2,12 @@ package block

import (
"context"
"fmt"
"time"
)

// AggregationLoop is responsible for aggregating transactions into rollup-blocks.
func (m *Manager) AggregationLoop(ctx context.Context) {
func (m *Manager) AggregationLoop(ctx context.Context, errCh chan<- error) {
initialHeight := m.genesis.InitialHeight //nolint:gosec
height, err := m.store.Height(ctx)
if err != nil {
Expand DownExpand Up@@ -37,30 +38,39 @@ func (m *Manager) AggregationLoop(ctx context.Context) {
// In Lazy Aggregator mode, blocks are built only when there are
// transactions or every LazyBlockTime.
if m.config.Node.LazyMode {
m.lazyAggregationLoop(ctx, blockTimer)
if err := m.lazyAggregationLoop(ctx, blockTimer); err != nil {
errCh <- fmt.Errorf("error in lazy aggregation loop: %w", err)
}
return
}

m.normalAggregationLoop(ctx, blockTimer)
if err := m.normalAggregationLoop(ctx, blockTimer); err != nil {
errCh <- fmt.Errorf("error in normal aggregation loop: %w", err)
}
}

func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Timer) {
func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Timer) error {
// lazyTimer triggers block publication even during inactivity
lazyTimer := time.NewTimer(0)
defer lazyTimer.Stop()

for {
select {
case <-ctx.Done():
return
return nil

case <-lazyTimer.C:
m.logger.Debug("Lazy timer triggered block production")
m.produceBlock(ctx, "lazy_timer", lazyTimer, blockTimer)

if err := m.produceBlock(ctx, "lazy_timer", lazyTimer, blockTimer); err != nil {
return err
}
Comment on lines +65 to +67

@coderabbitaicoderabbitaiBotMay 9, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Loop exits on single publish error

Returning the error here causes the entire lazy loop to stop after a single failed publishBlock.
If the goal is “halt at app failure” but still tolerate transient DA/network errors, consider distinguishing between fatal and retryable errors, e.g.:

iferr:= m.produceBlock(...); err!=nil {
iferrors.Is(err, coreexecutor.ErrAppError) { // fatalreturnerr// bubble & panic
}
m.logger.Error("transient publish error", "err", err)
continue// retry on next tick
}

case <-blockTimer.C:
if m.txsAvailable {
m.produceBlock(ctx, "block_timer", lazyTimer, blockTimer)
if err := m.produceBlock(ctx, "block_timer", lazyTimer, blockTimer); err != nil {
return err
}

m.txsAvailable = false
} else {
// Ensure we keep ticking even when there are no txs
Expand All@@ -73,34 +83,36 @@ func (m *Manager) lazyAggregationLoop(ctx context.Context, blockTimer *time.Time
}

// produceBlock handles the common logic for producing a block and resetting timers
func (m *Manager) produceBlock(ctx context.Context, mode string, lazyTimer, blockTimer *time.Timer) {
// Record the start time
func (m *Manager) produceBlock(ctx context.Context, mode string, lazyTimer, blockTimer *time.Timer) error {
start := time.Now()

// Attempt to publish the block
if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
m.logger.Error("error while publishing block", "mode", mode, "error", err)
} else {
m.logger.Debug("Successfully published block", "mode", mode)
return fmt.Errorf("error while publishing block: %w", err)
}

m.logger.Debug("Successfully published block", "mode", mode)

// Reset both timers for the next aggregation window
lazyTimer.Reset(getRemainingSleep(start, m.config.Node.LazyBlockInterval.Duration))
blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))

return nil
}

func (m *Manager) normalAggregationLoop(ctx context.Context, blockTimer *time.Timer) {
func (m *Manager) normalAggregationLoop(ctx context.Context, blockTimer *time.Timer) error {
for {
select {
case <-ctx.Done():
return
return nil
case <-blockTimer.C:
// Define the start time for the block production period
start := time.Now()

if err := m.publishBlock(ctx); err != nil && ctx.Err() == nil {
m.logger.Error("error while publishing block", "error", err)
return fmt.Errorf("error while publishing block: %w", err)
}

// Reset the blockTimer to signal the next block production
// period based on the block time.
blockTimer.Reset(getRemainingSleep(start, m.config.Node.BlockTime.Duration))
Expand Down
19 changes: 7 additions & 12 deletions block/aggregation_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -82,7 +82,7 @@ func TestAggregationLoop_Normal_BasicInterval(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
m.AggregationLoop(ctx)
m.AggregationLoop(ctx, make(chan<- error))
m.logger.Info("AggregationLoop exited")
}()

Expand DownExpand Up@@ -118,12 +118,10 @@ func TestAggregationLoop_Normal_BasicInterval(t *testing.T) {
// TestAggregationLoop_Normal_PublishBlockError verifies that the aggregation loop handles errors from publishBlock gracefully.
func TestAggregationLoop_Normal_PublishBlockError(t *testing.T) {
t.Parallel()
assert := assert.New(t)
require := require.New(t)

blockTime := 50 * time.Millisecond
waitTime := blockTime*4 + blockTime/2
tolerance := blockTime / 2

mockStore := mocks.NewStore(t)
mockStore.On("Height", mock.Anything).Return(uint64(1), nil).Maybe()
Expand DownExpand Up@@ -185,10 +183,12 @@ func TestAggregationLoop_Normal_PublishBlockError(t *testing.T) {

ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
errCh := make(chan error, 1)

wg.Add(1)
go func() {
defer wg.Done()
m.AggregationLoop(ctx)
m.AggregationLoop(ctx, errCh)
m.logger.Info("AggregationLoop exited")
}()

Expand All@@ -201,12 +201,7 @@ func TestAggregationLoop_Normal_PublishBlockError(t *testing.T) {
defer publishLock.Unlock()

calls := publishCalls.Load()
assert.GreaterOrEqualf(calls, int64(4), "publishBlock should have been called multiple times (around 4), but was called %d times", calls)
assert.LessOrEqualf(calls, int64(5), "publishBlock should have been called multiple times (around 4-5), but was called %d times", calls)

require.GreaterOrEqual(len(publishTimes), 3, "Need at least 3 timestamps to check intervals after error")
for i := 2; i < len(publishTimes); i++ {
interval := publishTimes[i].Sub(publishTimes[i-1])
WithinDuration(t, blockTime, interval, tolerance)
}
require.Equal(calls, int64(1))
require.ErrorContains(<-errCh, expectedErr.Error())
require.Equal(len(publishTimes), 1, "Expected only one publish time after error")

@julienrbrtjulienrbrtMay 20, 2025

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The relevant publishing issues that shouldn't halt have been changed to warning logs.

}
6 changes: 4 additions & 2 deletions block/daIncluder.go → block/da_includer.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -8,7 +8,7 @@ import (

// DAIncluderLoop is responsible for advancing the DAIncludedHeight by checking if blocks after the current height
// have both their header and data marked as DA-included in the caches. If so, it calls setDAIncludedHeight.
func (m *Manager) DAIncluderLoop(ctx context.Context) {
func (m *Manager) DAIncluderLoop(ctx context.Context, errCh chan<- error) {
for {
select {
case <-ctx.Done():
Expand All@@ -28,8 +28,10 @@ func (m *Manager) DAIncluderLoop(ctx context.Context) {
if daIncluded {
// Both header and data are DA-included, so we can advance the height
if err := m.incrementDAIncludedHeight(ctx); err != nil {
panic(fmt.Errorf("error while incrementing DA included height: %w", err))
errCh <- fmt.Errorf("error while incrementing DA included height: %w", err)
return
}

currentDAIncluded = nextHeight
} else {
// Stop at the first block that is not DA-included
Expand Down
14 changes: 7 additions & 7 deletions block/daIncluder_test.go → block/da_includer_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -67,7 +67,7 @@ func TestDAIncluderLoop_AdvancesHeightWhenBothDAIncluded(t *testing.T) {

go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -100,7 +100,7 @@ func TestDAIncluderLoop_StopsWhenHeaderNotDAIncluded(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -132,7 +132,7 @@ func TestDAIncluderLoop_StopsWhenDataNotDAIncluded(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -163,7 +163,7 @@ func TestDAIncluderLoop_StopsOnGetBlockDataError(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -286,7 +286,7 @@ func TestDAIncluderLoop_MultipleConsecutiveHeightsDAIncluded(t *testing.T) {
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -325,7 +325,7 @@ func TestDAIncluderLoop_AdvancesHeightWhenDataHashIsEmptyAndHeaderDAIncluded(t *
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand DownExpand Up@@ -357,7 +357,7 @@ func TestDAIncluderLoop_DoesNotAdvanceWhenDataHashIsEmptyAndHeaderNotDAIncluded(
wg.Add(1)
go func() {
defer wg.Done()
m.DAIncluderLoop(ctx)
m.DAIncluderLoop(ctx, make(chan<- error))
}()

m.sendNonBlockingSignalToDAIncluderCh()
Expand Down
5 changes: 3 additions & 2 deletions block/da_speed_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -78,7 +78,7 @@ func TestDASpeed(t *testing.T) {
go manager.RetrieveLoop(ctx)
go manager.HeaderStoreRetrieveLoop(ctx)
go manager.DataStoreRetrieveLoop(ctx)
go manager.SyncLoop(ctx)
go manager.SyncLoop(ctx, make(chan<- error))

// then
assert.Eventually(t, func() bool {
Expand DownExpand Up@@ -117,7 +117,8 @@ func setupManagerForTest(t *testing.T, initialDAHeight uint64) (*Manager, *rollm
store: mockStore,
config: config.Config{
Node: config.NodeConfig{BlockTime: config.DurationWrapper{Duration: blockTime}},
DA: config.DAConfig{BlockTime: config.DurationWrapper{Duration: blockTime}}},
DA: config.DAConfig{BlockTime: config.DurationWrapper{Duration: blockTime}},
},
genesis: genesis.Genesis{ProposerAddress: addr},
daHeight: new(atomic.Uint64),
headerInCh: make(chan NewHeaderEvent),
Expand Down
14 changes: 0 additions & 14 deletions block/errors.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2,7 +2,6 @@ package block

import (
"errors"
"fmt"
)

// These errors are used by Manager.
Expand All@@ -19,16 +18,3 @@ var (
// ErrHeightFromFutureStr is the error message for height from future returned by da
ErrHeightFromFutureStr = errors.New("given height is from the future")
)

// SaveBlockError is returned on failure to save block data
type SaveBlockError struct {
Err error
}

func (e SaveBlockError) Error() string {
return fmt.Sprintf("failed to save block: %v", e.Err)
}

func (e SaveBlockError) Unwrap() error {
return e.Err
}
30 changes: 9 additions & 21 deletions block/lazy_aggregation_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ func TestLazyAggregationLoop_BlockTimerTrigger(t *testing.T) {
defer wg.Done()
blockTimer := time.NewTimer(0) // Fire immediately first time
defer blockTimer.Stop()
m.lazyAggregationLoop(ctx, blockTimer)
require.NoError(m.lazyAggregationLoop(ctx, blockTimer))
}()

// Wait for at least one block to be published
Expand DownExpand Up@@ -137,7 +137,7 @@ func TestLazyAggregationLoop_LazyTimerTrigger(t *testing.T) {
// Use real timers for this test
blockTimer := time.NewTimer(0) // Fire immediately first time
defer blockTimer.Stop()
m.lazyAggregationLoop(ctx, blockTimer)
require.NoError(m.lazyAggregationLoop(ctx, blockTimer))
}()

// Wait for the first publish call triggered by the initial immediate lazyTimer fire
Expand All@@ -163,7 +163,7 @@ func TestLazyAggregationLoop_LazyTimerTrigger(t *testing.T) {
wg.Wait()
}

// TestLazyAggregationLoop_PublishError tests that the loop continues after a publish error.
// TestLazyAggregationLoop_PublishError tests that the loop exits.
func TestLazyAggregationLoop_PublishError(t *testing.T) {
t.Parallel()
require := require.New(t)
Expand All@@ -176,7 +176,7 @@ func TestLazyAggregationLoop_PublishError(t *testing.T) {
pubMock.err = errors.New("publish failed")
pubMock.mu.Unlock()

ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithCancel(t.Context())
defer cancel()

var wg sync.WaitGroup
Expand All@@ -186,7 +186,7 @@ func TestLazyAggregationLoop_PublishError(t *testing.T) {
// Use real timers
blockTimer := time.NewTimer(0)
defer blockTimer.Stop()
m.lazyAggregationLoop(ctx, blockTimer)
require.Error(m.lazyAggregationLoop(ctx, blockTimer))
}()

// Wait for the first publish attempt (which will fail)
Expand All@@ -196,20 +196,8 @@ func TestLazyAggregationLoop_PublishError(t *testing.T) {
require.Fail("timed out waiting for first block publication attempt")
}

// Remove the error for subsequent calls
pubMock.mu.Lock()
pubMock.err = nil
pubMock.mu.Unlock()

// Wait for the second publish attempt (should succeed)
// Use a longer timeout since we need to wait for either the lazy timer or block timer to fire
select {
case <-pubMock.calls:
case <-time.After(2 * lazyTime): // Use the longer of the two timers with some buffer
require.Fail("timed out waiting for second block publication attempt after error")
}
// loop exited, nothing to do.

cancel()
wg.Wait()
}

Expand DownExpand Up@@ -267,7 +255,7 @@ func TestLazyAggregationLoop_TxNotification(t *testing.T) {
// Start with a timer that won't fire immediately
blockTimer := time.NewTimer(blockTime)
defer blockTimer.Stop()
m.lazyAggregationLoop(ctx, blockTimer)
require.NoError(m.lazyAggregationLoop(ctx, blockTimer))
}()

// Wait for the initial lazy timer to fire and publish a block
Expand DownExpand Up@@ -341,7 +329,7 @@ func TestEmptyBlockCreation(t *testing.T) {
defer blockTimer.Stop()

// Call produceBlock directly to test empty block creation
m.produceBlock(ctx, "test_trigger", lazyTimer, blockTimer)
require.NoError(m.produceBlock(ctx, "test_trigger", lazyTimer, blockTimer))

// Verify that the context was passed correctly
require.NotNil(capturedCtx, "Context should have been captured by mock publish function")
Expand DownExpand Up@@ -369,7 +357,7 @@ func TestNormalAggregationLoop_TxNotification(t *testing.T) {
defer wg.Done()
blockTimer := time.NewTimer(blockTime)
defer blockTimer.Stop()
m.normalAggregationLoop(ctx, blockTimer)
require.NoError(m.normalAggregationLoop(ctx, blockTimer))
}()

// Wait for the first block to be published by the timer
Expand Down
Loading