Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
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: 28 additions & 14 deletions block/aggregation.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package block

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

Expand Down Expand Up @@ -37,30 +38,41 @@ 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 {
panic(fmt.Errorf("error in lazy aggregation loop: %w", err))
}

return
}

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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This change assumes all errors from publishBlock are app related failures.

If we bubble up all the errors up from publishBlock and panic, it's also going to panic on when maxPendingHeaders reaches its limit which is not an app failure. We should NOT panic on this.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Ok, I'll update the non error as warning logs and return without an error.
IMHO, we should only return an error if we need to stop

}
}

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)

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

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.:

if err := m.produceBlock(...); err != nil {
    if errors.Is(err, coreexecutor.ErrAppError) { // fatal
        return err     // 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 +85,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
File renamed without changes.
File renamed without changes.
14 changes: 0 additions & 14 deletions block/errors.go
Original file line number Diff line number Diff 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
}
12 changes: 6 additions & 6 deletions block/lazy_aggregation_test.go
Original file line number Diff line number Diff 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 Down Expand 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 Down Expand Up @@ -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 Down Expand Up @@ -267,7 +267,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 Down Expand Up @@ -342,7 +342,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 Down Expand Up @@ -370,7 +370,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
44 changes: 21 additions & 23 deletions block/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -175,8 +175,7 @@ func getInitialState(ctx context.Context, genesis genesis.Genesis, signer signer
// TODO(tzdybal): handle max bytes
stateRoot, _, err := exec.InitChain(ctx, genesis.GenesisDAStartTime, genesis.InitialHeight, genesis.ChainID)
if err != nil {
logger.Error("error while initializing chain", "error", err)
return types.State{}, err
return types.State{}, fmt.Errorf("failed to initialize chain: %w", err)
}

// Initialize genesis block explicitly
Expand All @@ -188,7 +187,8 @@ func getInitialState(ctx context.Context, genesis genesis.Genesis, signer signer
ChainID: genesis.ChainID,
Height: genesis.InitialHeight,
Time: uint64(genesis.GenesisDAStartTime.UnixNano()),
}}
},
}

var signature types.Signature

Expand Down Expand Up @@ -272,10 +272,10 @@ func NewManager(
) (*Manager, error) {
s, err := getInitialState(ctx, genesis, signer, store, exec, logger)
if err != nil {
logger.Error("error while getting initial state", "error", err)
return nil, err
return nil, fmt.Errorf("failed to get initial state: %w", err)
}
//set block height in store

// set block height in store
err = store.SetHeight(ctx, s.LastBlockHeight)
if err != nil {
return nil, err
Expand Down Expand Up @@ -507,8 +507,8 @@ func (m *Manager) publishBlockInternal(ctx context.Context) error {
}

if m.config.Node.MaxPendingHeaders != 0 && m.pendingHeaders.numPendingHeaders() >= m.config.Node.MaxPendingHeaders {
return fmt.Errorf("refusing to create block: pending blocks [%d] reached limit [%d]",
m.pendingHeaders.numPendingHeaders(), m.config.Node.MaxPendingHeaders)
m.logger.Warn("refusing to create block: pending blocks [%d] reached limit [%d]", m.pendingHeaders.numPendingHeaders(), m.config.Node.MaxPendingHeaders)
return nil
}

var (
Expand Down Expand Up @@ -566,7 +566,8 @@ func (m *Manager) publishBlockInternal(ctx context.Context) error {
}
m.logger.Info("Creating empty block", "height", newHeight)
} else {
return fmt.Errorf("failed to get transactions from batch: %w", err)
m.logger.Warn("failed to get transactions from batch", "error", err)
return nil
}
} else {
if batchData.Before(lastHeaderTime) {
Expand All @@ -582,7 +583,7 @@ func (m *Manager) publishBlockInternal(ctx context.Context) error {
}

if err = m.store.SaveBlockData(ctx, header, data, &signature); err != nil {
return SaveBlockError{err}
return fmt.Errorf("failed to save block: %w", err)
}
}

Expand All @@ -596,16 +597,12 @@ func (m *Manager) publishBlockInternal(ctx context.Context) error {

if err := header.ValidateBasic(); err != nil {
// If this ever happens, for recovery, check for a mismatch between the configured signing key and the proposer address in the genesis file
panic(fmt.Errorf("critical: newly produced header failed validation: %w", err))
return fmt.Errorf("header validation error: %w", err)
}

newState, err := m.applyBlock(ctx, header, data)
if err != nil {
if ctx.Err() != nil {
return err
}
// if call to applyBlock fails, we halt the node, see https://github.com/cometbft/cometbft/pull/496
panic(err)
return fmt.Errorf("error applying block: %w", err)
}

// append metadata to Data before validating and saving
Expand All @@ -628,23 +625,23 @@ func (m *Manager) publishBlockInternal(ctx context.Context) error {
// SaveBlock commits the DB tx
err = m.store.SaveBlockData(ctx, header, data, &signature)
if err != nil {
return SaveBlockError{err}
return fmt.Errorf("failed to save block: %w", err)
}

// Update the store height before submitting to the DA layer but after committing to the DB
err = m.store.SetHeight(ctx, headerHeight)
if err != nil {
if err = m.store.SetHeight(ctx, headerHeight); err != nil {
return err
}

newState.DAHeight = m.daHeight.Load()
// After this call m.lastState is the NEW state returned from ApplyBlock
// updateState also commits the DB tx
err = m.updateState(ctx, newState)
if err != nil {
return err
if err = m.updateState(ctx, newState); err != nil {
return fmt.Errorf("failed to update state: %w", err)
}

m.recordMetrics(data)

// Check for shut down event prior to sending the header and block to
// their respective channels. The reason for checking for the shutdown
// event separately is due to the inconsistent nature of the select
Expand Down Expand Up @@ -820,9 +817,10 @@ func (m *Manager) execApplyBlock(ctx context.Context, lastState types.State, hea
for i := range data.Txs {
rawTxs[i] = data.Txs[i]
}

newStateRoot, _, err := m.exec.ExecuteTxs(ctx, rawTxs, header.Height(), header.Time(), lastState.AppHash)
if err != nil {
return types.State{}, err
return types.State{}, fmt.Errorf("failed to execute transactions: %w", err)
}

s, err := lastState.NextState(header, newStateRoot)
Expand Down
2 changes: 1 addition & 1 deletion block/submitter.go
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ daSubmitRetryLoop:
}
}

ctx, cancel := context.WithTimeout(ctx, 60*time.Second) //TODO: make this configurable
ctx, cancel := context.WithTimeout(ctx, 60*time.Second) // TODO: make this configurable
res := types.SubmitWithHelpers(ctx, m.da, m.logger, headersBz, gasPrice, nil)
cancel()

Expand Down
36 changes: 15 additions & 21 deletions block/sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,7 @@ import (

// SyncLoop is responsible for syncing blocks.
//
// SyncLoop processes headers gossiped in P2P network to know what's the latest block height,
// block data is retrieved from DA layer.
// SyncLoop processes headers gossiped in P2P network to know what's the latest block height, block data is retrieved from DA layer.
func (m *Manager) SyncLoop(ctx context.Context) {
daTicker := time.NewTicker(m.config.DA.BlockTime.Duration)
defer daTicker.Stop()
Expand Down Expand Up @@ -63,11 +62,10 @@ func (m *Manager) SyncLoop(ctx context.Context) {
}
}

err = m.trySyncNextBlock(ctx, daHeight)
if err != nil {
m.logger.Info("failed to sync next block", "error", err)
continue
if err = m.trySyncNextBlock(ctx, daHeight); err != nil {
panic(fmt.Errorf("failed to sync next block: %w", err))
}
Comment thread
julienrbrt marked this conversation as resolved.
Comment on lines +65 to 68

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

⚠️ Potential issue

Guard against context.Canceled and avoid a potential sender-block on errCh

  1. trySyncNextBlock returns ctx.Err() when the node is shutting down.
    Forwarding a plain context.Canceled into errCh forces the whole node into the error branch even though this is an expected shutdown path.

  2. errCh is created with capacity 1. If another goroutine already sent an error that hasn’t been drained yet, this send will block forever, freezing the sync goroutine.

A defensive pattern is:

-if err = m.trySyncNextBlock(ctx, daHeight); err != nil {
-    errCh <- fmt.Errorf("failed to sync next block: %w", err)
-    return
-}
+if err = m.trySyncNextBlock(ctx, daHeight); err != nil &&
+   !errors.Is(err, context.Canceled) {
+    select {
+    case errCh <- fmt.Errorf("failed to sync next block: %w", err):
+    default: // channel already holds an error – don’t block
+    }
+    return
+}

You’ll need to add errors to the import list.

🤖 Prompt for AI Agents
In block/sync.go around lines 65 to 68, the current code sends errors including
context.Canceled directly to errCh, which causes unnecessary error handling on
expected shutdowns and can block if errCh is full. Modify the code to check if
the error is context.Canceled and skip sending it to errCh in that case. Also,
use a non-blocking send to errCh by selecting on errCh and a default case to
avoid blocking if errCh is already full. Add the "errors" package to imports to
use errors.Is for error comparison.


m.headerCache.SetSeen(headerHash)
case dataEvent := <-m.dataInCh:
data := dataEvent.Data
Expand Down Expand Up @@ -115,8 +113,7 @@ func (m *Manager) SyncLoop(ctx context.Context) {

err = m.trySyncNextBlock(ctx, daHeight)
if err != nil {
m.logger.Info("failed to sync next block", "error", err)
continue
panic(fmt.Errorf("failed to sync next block: %w", err))
}
Comment on lines 114 to 119

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

Replicate the shutdown-friendly panic filter here as well

Mirror the context.Canceled guard suggested for the header branch to keep behaviour consistent.

-if err != nil {
-    panic(fmt.Errorf("failed to sync next block: %w", err))
-}
+if err != nil && !errors.Is(err, context.Canceled) {
+    panic(fmt.Errorf("failed to sync next block: %w", err))
+}
🤖 Prompt for AI Agents
In block/sync.go around lines 114 to 117, the panic on error from
trySyncNextBlock should be guarded against context.Canceled errors to allow
graceful shutdown. Modify the error handling to check if the error is
context.Canceled and skip panicking in that case, mirroring the
shutdown-friendly panic filter used in the header branch. This ensures
consistent behavior during shutdown scenarios.

m.dataCache.SetSeen(dataHash)
case <-ctx.Done():
Expand Down Expand Up @@ -159,32 +156,29 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {
if err := m.Validate(ctx, h, d); err != nil {
return fmt.Errorf("failed to validate block: %w", err)
}

newState, err := m.applyBlock(ctx, h, d)
if err != nil {
if ctx.Err() != nil {
return err
}
// if call to applyBlock fails, we halt the node, see https://github.com/cometbft/cometbft/pull/496
panic(fmt.Errorf("failed to ApplyBlock: %w", err))
return fmt.Errorf("failed to apply block: %w", err)
}
err = m.store.SaveBlockData(ctx, h, d, &h.Signature)
if err != nil {
return SaveBlockError{err}

if err = m.store.SaveBlockData(ctx, h, d, &h.Signature); err != nil {
return fmt.Errorf("failed to save block: %w", err)
}

// Height gets updated
err = m.store.SetHeight(ctx, hHeight)
if err != nil {
if err = m.store.SetHeight(ctx, hHeight); err != nil {
return err
}

if daHeight > newState.DAHeight {
newState.DAHeight = daHeight
}
err = m.updateState(ctx, newState)
if err != nil {
m.logger.Error("failed to save updated state", "error", err)

if err = m.updateState(ctx, newState); err != nil {
return fmt.Errorf("failed to save updated state: %w", err)
}

m.headerCache.DeleteItem(currentHeight + 1)
m.dataCache.DeleteItem(currentHeight + 1)
m.dataCache.DeleteItemByHash(h.DataHash.String())
Expand Down