Repository navigation
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.
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
Changes from 2 commits
5e77da5
d4d6fc3
1f75ad7
341a37b
c6a785f
900fb06
e5dcf6b
3db6aca
5cdd35f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,6 +2,7 @@ package block | |
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
| ) | ||
|
|
||
|
|
@@ -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)) | ||
| } | ||
| } | ||
|
|
||
| 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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 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 | ||
|
|
@@ -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)) | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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() | ||
|
|
@@ -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)) | ||
| } | ||
|
julienrbrt marked this conversation as resolved.
Comment on lines
+65
to
68
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🛠️ Refactor suggestion Guard against
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 🤖 Prompt for AI Agents |
||
|
|
||
| m.headerCache.SetSeen(headerHash) | ||
| case dataEvent := <-m.dataInCh: | ||
| data := dataEvent.Data | ||
|
|
@@ -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
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 -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 |
||
| m.dataCache.SetSeen(dataHash) | ||
| case <-ctx.Done(): | ||
|
|
@@ -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()) | ||
|
|
||
There was a problem hiding this comment.
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
publishBlockand panic, it's also going to panic on whenmaxPendingHeadersreaches its limit which is not an app failure. We should NOT panic on this.There was a problem hiding this comment.
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