Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 276
refactor!: syncing #2798
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
Merged
Uh oh!
There was an error while loading. Please reload this page.
Merged
refactor!: syncing #2798
Changes from all commits
Commits
Show all changes
29 commits
Select commit
Hold shift + click to select a range
a679cc4
fix(block/internal/syncing): only publish header/data locally if sour…
renaynay 28aa9f6
generate mocks
tac0turtle 40d6abc
Merge branch 'main' into rene-tryfix
tac0turtle 2d9cf4d
Merge branch 'main' into rene-tryfix
tac0turtle 46df9ea
fix serlization
tac0turtle 1463345
refactor
tac0turtle 94140e3
refactor syncer
tac0turtle a8963d4
make goheader event driven
tac0turtle cb4f6d7
feedback
tac0turtle c60a425
simplify p2phandler due to event driven design
tac0turtle ade638e
inline functions in p2p_handler.go
tac0turtle e5a67bb
remove redundant call
tac0turtle 626e3de
one file
tac0turtle 3a5488f
lint
tac0turtle c62d42d
remove hex
tac0turtle aa1b5a2
Create adr-023-event-driven-sync.md
tac0turtle 7056f74
Update adr-023-event-driven-sync.md
tac0turtle fc9d17f
Merge branch 'main' into marko/p2p_refactor
tac0turtle b3f0a90
build and run scripts
tac0turtle 4cf2a07
refactor: try CN approach (#2812)
tac0turtle c12978a
Merge branch 'main' into marko/p2p_refactor
tac0turtle 0c9ac5a
update comments
tac0turtle 9673004
change da sync flow
tac0turtle 699cadf
add timeout for da request
tac0turtle 8d12323
fix errors
tac0turtle c2f8e71
remove replace
tac0turtle 8e73552
dont block on empty events as this means we got the response but there
tac0turtle 78d375c
remove comment and update debug log
tac0turtle e55fa1d
add comment back
tac0turtle File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -3,9 +3,8 @@ package syncing | ||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "time" | ||
| "sync/atomic" | ||
| goheader "github.com/celestiaorg/go-header" | ||
| "github.com/rs/zerolog" | ||
| @@ -16,16 +15,23 @@ import ( | ||
| "github.com/evstack/ev-node/types" | ||
| ) | ||
| // P2PHandler handles all P2P operations for the syncer | ||
| // P2PHandler coordinates block retrieval from P2P stores for the syncer. | ||
| // It waits for both header and data to be available at a given height, | ||
| // validates their consistency, and emits events to the syncer for processing. | ||
| // | ||
| // The handler maintains a processedHeight to track the highest block that has been | ||
| // successfully validated and sent to the syncer, preventing duplicate processing. | ||
| type P2PHandler struct { | ||
| headerStore goheader.Store[*types.SignedHeader] | ||
| dataStore goheader.Store[*types.Data] | ||
| cache cache.Manager | ||
| genesis genesis.Genesis | ||
| logger zerolog.Logger | ||
| processedHeight atomic.Uint64 | ||
| } | ||
| // NewP2PHandler creates a new P2P handler | ||
| // NewP2PHandler creates a new P2P handler. | ||
| func NewP2PHandler( | ||
| headerStore goheader.Store[*types.SignedHeader], | ||
| dataStore goheader.Store[*types.Data], | ||
| @@ -42,176 +48,76 @@ func NewP2PHandler( | ||
| } | ||
| } | ||
| // ProcessHeaderRange processes headers from the header store within the given range | ||
| func (h *P2PHandler) ProcessHeaderRange(ctx context.Context, startHeight, endHeight uint64, heightInCh chan<- common.DAHeightEvent) { | ||
| if startHeight > endHeight { | ||
| return | ||
| } | ||
| for height := startHeight; height <= endHeight; height++ { | ||
| select { | ||
| case <-ctx.Done(): | ||
| // SetProcessedHeight updates the highest processed block height. | ||
| func (h *P2PHandler) SetProcessedHeight(height uint64) { | ||
| for { | ||
| current := h.processedHeight.Load() | ||
| if height <= current { | ||
| return | ||
| default: | ||
| } | ||
| // Create a timeout context for each GetByHeight call to prevent blocking | ||
| timeoutCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) | ||
| header, err := h.headerStore.GetByHeight(timeoutCtx, height) | ||
| cancel() | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) { | ||
| h.logger.Debug().Uint64("height", height).Msg("timeout waiting for header from store, will retry later") | ||
| // Don't continue processing further heights if we timeout on one | ||
| // This prevents blocking on sequential heights | ||
| return | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("failed to get header from store") | ||
| continue | ||
| } | ||
| // basic header validation | ||
| if err := h.assertExpectedProposer(header.ProposerAddress); err != nil { | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P") | ||
| continue | ||
| } | ||
| // Get corresponding data (empty data are still broadcasted by peers) | ||
| var data *types.Data | ||
| timeoutCtx, cancel = context.WithTimeout(ctx, 500*time.Millisecond) | ||
| retrievedData, err := h.dataStore.GetByHeight(timeoutCtx, height) | ||
| cancel() | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) { | ||
| h.logger.Debug().Uint64("height", height).Msg("timeout waiting for data from store, will retry later") | ||
| // Don't continue processing if data is not available | ||
| // Store event with header only for later processing | ||
| continue | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("could not retrieve data for header from data store") | ||
| continue | ||
| } | ||
| data = retrievedData | ||
| // CRITICAL: Validate that data matches the header's DataHash commitment | ||
| // This prevents accepting legitimate headers paired with tampered data from different blocks | ||
| dataCommitment := data.DACommitment() | ||
| if !bytes.Equal(header.DataHash[:], dataCommitment[:]) { | ||
| h.logger.Warn(). | ||
| Uint64("height", height). | ||
| Str("header_data_hash", fmt.Sprintf("%x", header.DataHash)). | ||
| Str("actual_data_hash", fmt.Sprintf("%x", dataCommitment)). | ||
| Msg("DataHash mismatch: header and data do not match from P2P, discarding") | ||
| continue | ||
| } | ||
| // further header validation (signature) is done in validateBlock. | ||
| // we need to be sure that the previous block n-1 was executed before validating block n | ||
| // Create height event | ||
| event := common.DAHeightEvent{ | ||
| Header: header, | ||
| Data: data, | ||
| DaHeight: 0, // P2P events don't have DA height context | ||
| Source: common.SourceP2P, | ||
| } | ||
| select { | ||
| case heightInCh <- event: | ||
| default: | ||
| h.cache.SetPendingEvent(event.Header.Height(), &event) | ||
| if h.processedHeight.CompareAndSwap(current, height) { | ||
| return | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Str("source", "p2p_headers").Msg("processed header from P2P") | ||
| } | ||
| } | ||
| // ProcessDataRange processes data from the data store within the given range | ||
| func (h *P2PHandler) ProcessDataRange(ctx context.Context, startHeight, endHeight uint64, heightInCh chan<- common.DAHeightEvent) { | ||
| if startHeight > endHeight { | ||
| return | ||
| // ProcessHeight retrieves and validates both header and data for the given height from P2P stores. | ||
| // It blocks until both are available, validates consistency (proposer address and data hash match), | ||
| // then emits the event to heightInCh or stores it as pending. Updates processedHeight on success. | ||
| func (h *P2PHandler) ProcessHeight(ctx context.Context, height uint64, heightInCh chan<- common.DAHeightEvent) error { | ||
| if height <= h.processedHeight.Load() { | ||
| return nil | ||
| } | ||
| for height := startHeight; height <= endHeight; height++ { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| default: | ||
| } | ||
| // Create a timeout context for each GetByHeight call to prevent blocking | ||
| timeoutCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) | ||
| data, err := h.dataStore.GetByHeight(timeoutCtx, height) | ||
| cancel() | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) { | ||
| h.logger.Debug().Uint64("height", height).Msg("timeout waiting for data from store, will retry later") | ||
| // Don't continue processing further heights if we timeout on one | ||
| // This prevents blocking on sequential heights | ||
| return | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("failed to get data from store") | ||
| continue | ||
| } | ||
| // Get corresponding header with timeout | ||
| timeoutCtx, cancel = context.WithTimeout(ctx, 500*time.Millisecond) | ||
| header, err := h.headerStore.GetByHeight(timeoutCtx, height) | ||
| cancel() | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) { | ||
| h.logger.Debug().Uint64("height", height).Msg("timeout waiting for header from store, will retry later") | ||
| // Don't continue processing if header is not available | ||
| continue | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("could not retrieve header for data from header store") | ||
| continue | ||
| header, err := h.headerStore.GetByHeight(ctx, height) | ||
| if err != nil { | ||
| if ctx.Err() == nil { | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("header unavailable in store") | ||
| } | ||
| return err | ||
| } | ||
| if err := h.assertExpectedProposer(header.ProposerAddress); err != nil { | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P") | ||
| return err | ||
| } | ||
| // basic header validation | ||
| if err := h.assertExpectedProposer(header.ProposerAddress); err != nil { | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("invalid header from P2P") | ||
| continue | ||
| data, err := h.dataStore.GetByHeight(ctx, height) | ||
| if err != nil { | ||
| if ctx.Err() == nil { | ||
| h.logger.Debug().Uint64("height", height).Err(err).Msg("data unavailable in store") | ||
| } | ||
| return err | ||
| } | ||
| // CRITICAL: Validate that data matches the header's DataHash commitment | ||
| // This prevents accepting legitimate headers paired with tampered data from different blocks | ||
| dataCommitment := data.DACommitment() | ||
| if !bytes.Equal(header.DataHash[:], dataCommitment[:]) { | ||
| h.logger.Warn(). | ||
| Uint64("height", height). | ||
| Str("header_data_hash", fmt.Sprintf("%x", header.DataHash)). | ||
| Str("actual_data_hash", fmt.Sprintf("%x", dataCommitment)). | ||
| Msg("DataHash mismatch: header and data do not match from P2P, discarding") | ||
| continue | ||
| } | ||
| dataCommitment := data.DACommitment() | ||
| if !bytes.Equal(header.DataHash[:], dataCommitment[:]) { | ||
| err := fmt.Errorf("data hash mismatch: header %x, data %x", header.DataHash, dataCommitment) | ||
| h.logger.Warn().Uint64("height", height).Err(err).Msg("discarding inconsistent block from P2P") | ||
| return err | ||
| } | ||
| // further header validation (signature) is done in validateBlock. | ||
julienrbrt marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| // we need to be sure that the previous block n-1 was executed before validating block n | ||
| // further header validation (signature) is done in validateBlock. | ||
| // we need to be sure that the previous block n-1 was executed before validating block n | ||
| event := common.DAHeightEvent{ | ||
| Header: header, | ||
| Data: data, | ||
| DaHeight: 0, | ||
| Source: common.SourceP2P, | ||
| } | ||
| // Create height event | ||
| event := common.DAHeightEvent{ | ||
| Header: header, | ||
| Data: data, | ||
| DaHeight: 0, // P2P events don't have DA height context | ||
| Source: common.SourceP2P, | ||
| } | ||
| select { | ||
| case heightInCh <- event: | ||
| default: | ||
| h.cache.SetPendingEvent(event.Header.Height(), &event) | ||
| } | ||
| select { | ||
| case heightInCh <- event: | ||
| default: | ||
| h.cache.SetPendingEvent(event.Header.Height(), &event) | ||
| } | ||
| h.SetProcessedHeight(height) | ||
| h.logger.Debug().Uint64("height", height).Str("source", "p2p_data").Msg("processed data from P2P") | ||
| } | ||
| h.logger.Debug().Uint64("height", height).Msg("processed event from P2P") | ||
| return nil | ||
| } | ||
| // assertExpectedProposer validates the proposer address | ||
| // assertExpectedProposer validates the proposer address. | ||
| func (h *P2PHandler) assertExpectedProposer(proposerAddr []byte) error { | ||
| if !bytes.Equal(h.genesis.ProposerAddress, proposerAddr) { | ||
| return fmt.Errorf("proposer address mismatch: got %x, expected %x", | ||
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.