migration boundry fix #288
Merged
Conversation
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
insight-indexer - insight-8453-baseWe're building your pull request over Zeet. We have provided links for Zeet apps linked to this github repo
|
WalkthroughIntroduces RPC-based latest-block retrieval for migration boundary calculations; adjusts end-block logic accordingly. Adds S3 storage configuration field for flush timeout. Updates S3 storage flush wait logic to use configurable timeout with a default applied when unset. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant CLI as CLI
participant Migrator as Migrator
participant RPC as RPC Node
participant Store as DB/Store
CLI->>Migrator: DetermineMigrationBoundaries(targetEndBlock)
Migrator->>RPC: GetLatestBlock()
alt RPC ok
RPC-->>Migrator: latestBlockRPC
Migrator->>Store: Read latestBlockStored
alt targetEndBlock == 0
Migrator-->>CLI: endBlock = latestBlockRPC
else targetEndBlock > 0
alt targetEndBlock <= latestBlockRPC
Migrator-->>CLI: endBlock = targetEndBlock
else targetEndBlock > latestBlockRPC
Migrator-->>CLI: endBlock = latestBlockStored (prior logic)
end
end
else RPC error
Migrator-->>CLI: Fatal log and exit
end
sequenceDiagram
autonumber
participant App as App
participant S3 as S3Storage
note over S3: On init: if cfg.FlushTimeout==0 → set 300s
App->>S3: TriggerFlush()
alt flush channel accepts
S3->>S3: wait on s.flushDoneCh within FlushTimeout
alt completed
S3-->>App: ok
else timeout
S3-->>App: error (timeout with configured seconds)
end
else channel full
S3->>S3: wait for completion within FlushTimeout
alt completed
S3-->>App: ok
else timeout
S3-->>App: error (timeout with configured seconds)
end
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Pre-merge checks (2 passed, 1 warning)❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
configs/config.go (1)
96-96: Align comment with runtime default (doc-runtime mismatch).The comment says default 60s, but S3 connector sets default to 300s. Align the docs to avoid confusion.
- FlushTimeout int `mapstructure:"flushTimeoutSeconds"` // Timeout in seconds for flush operations (default: 60) + FlushTimeout int `mapstructure:"flushTimeoutSeconds"` // Timeout in seconds for flush operations (default: 300)Optionally consider using time.Duration for timeout fields to make units explicit.
internal/storage/s3.go (1)
106-108: Default looks good; add basic bounds check.If a negative value is configured, the timeout becomes negative. Guard it to fall back to default.
- if cfg.FlushTimeout == 0 { + if cfg.FlushTimeout <= 0 { cfg.FlushTimeout = 300 // 5 mins default }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (3)
cmd/migrate_valid.go(1 hunks)configs/config.go(1 hunks)internal/storage/s3.go(2 hunks)
Comment on lines
+351
to
356
| latestBlockRPC, err := m.rpcClient.GetLatestBlockNumber(context.Background()) | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msg("Failed to get latest block from RPC") | ||
| } | ||
| log.Info().Msgf("Latest block in main storage: %d", latestBlockStored) | ||
|
|
There was a problem hiding this comment.
Fix log formatting for big.Int.
Using %d with *big.Int produces malformed output. Use %s or call .String().
- log.Info().Msgf("Latest block in main storage: %d", latestBlockStored)
+ log.Info().Msgf("Latest block in main storage: %s", latestBlockStored.String())📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| latestBlockRPC, err := m.rpcClient.GetLatestBlockNumber(context.Background()) | |
| if err != nil { | |
| log.Fatal().Err(err).Msg("Failed to get latest block from RPC") | |
| } | |
| log.Info().Msgf("Latest block in main storage: %d", latestBlockStored) | |
| latestBlockRPC, err := m.rpcClient.GetLatestBlockNumber(context.Background()) | |
| if err != nil { | |
| log.Fatal().Err(err).Msg("Failed to get latest block from RPC") | |
| } | |
| log.Info().Msgf("Latest block in main storage: %s", latestBlockStored.String()) |
🤖 Prompt for AI Agents
In cmd/migrate_valid.go around lines 351 to 356, the log uses fmt %d with a
*big.Int which yields malformed output; change the log to use %s with
latestBlockStored.String() (or call .String() and pass that) so the big.Int is
formatted correctly in the log message.
Comment on lines
+358
to
364
| if targetEndBlock.Sign() > 0 && targetEndBlock.Cmp(latestBlockRPC) <= 0 { | ||
| endBlock = targetEndBlock | ||
| } | ||
| if targetEndBlock.Uint64() == 0 { | ||
| endBlock = latestBlockRPC | ||
| } | ||
|
|
There was a problem hiding this comment.
End-boundary can exceed RPC latest when targetEndBlock > RPC; also missing start<=end guard.
Ensure we never migrate beyond RPC capability and validate the range.
- endBlock := latestBlockStored
- if targetEndBlock.Sign() > 0 && targetEndBlock.Cmp(latestBlockRPC) <= 0 {
- endBlock = targetEndBlock
- }
- if targetEndBlock.Uint64() == 0 {
- endBlock = latestBlockRPC
- }
+ // Cap by RPC latest, optionally by user-specified target
+ endBlock := new(big.Int).Set(latestBlockRPC)
+ if targetEndBlock.Sign() > 0 {
+ if targetEndBlock.Cmp(latestBlockRPC) <= 0 {
+ endBlock = targetEndBlock
+ } else {
+ // Requested end is beyond RPC; cap at RPC latest
+ endBlock = latestBlockRPC
+ }
+ }
+
+ // Validate final range
+ if endBlock.Cmp(targetStartBlock) < 0 {
+ log.Fatal().Msgf("Invalid migration range: end block %s is less than start block %s", endBlock.String(), targetStartBlock.String())
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if targetEndBlock.Sign() > 0 && targetEndBlock.Cmp(latestBlockRPC) <= 0 { | |
| endBlock = targetEndBlock | |
| } | |
| if targetEndBlock.Uint64() == 0 { | |
| endBlock = latestBlockRPC | |
| } | |
| // Cap by RPC latest, optionally by user-specified target | |
| endBlock := new(big.Int).Set(latestBlockRPC) | |
| if targetEndBlock.Sign() > 0 { | |
| if targetEndBlock.Cmp(latestBlockRPC) <= 0 { | |
| endBlock = targetEndBlock | |
| } else { | |
| // Requested end is beyond RPC; cap at RPC latest | |
| endBlock = latestBlockRPC | |
| } | |
| } | |
| // Validate final range | |
| if endBlock.Cmp(targetStartBlock) < 0 { | |
| log.Fatal().Msgf("Invalid migration range: end block %s is less than start block %s", endBlock.String(), targetStartBlock.String()) | |
| } |
🤖 Prompt for AI Agents
In cmd/migrate_valid.go around lines 358 to 364, the current logic can set
endBlock past the RPC latest and lacks a start<=end guard; change it to: if
targetEndBlock is zero use latestBlockRPC, else if targetEndBlock >
latestBlockRPC set endBlock = latestBlockRPC (cap to RPC), otherwise set
endBlock = targetEndBlock; after computing endBlock validate that
startBlock.Cmp(endBlock) <= 0 and return an error (or exit) if startBlock >
endBlock to prevent an invalid range.
Comment on lines
312
to
327
| select { | ||
| case <-s.flushDoneCh: | ||
| return nil | ||
| case <-time.After(60 * time.Second): | ||
| return fmt.Errorf("flush timeout after 60 seconds") | ||
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | ||
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | ||
| } | ||
| default: | ||
| // Flush channel is full, likely a flush is already in progress | ||
| // Wait for it to complete | ||
| select { | ||
| case <-s.flushDoneCh: | ||
| return nil | ||
| case <-time.After(60 * time.Second): | ||
| return fmt.Errorf("flush timeout after 60 seconds") | ||
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | ||
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Flush completion uses a shared unbuffered channel; concurrent Flush callers can time out or hang.
Only one waiter will receive from flushDoneCh per flush; others will block until another flush or hit timeout. Guard Flush with a mutex to serialize calls (minimal change), or switch to a broadcast/WaitGroup pattern.
Minimal mutex approach:
type S3Connector struct {
@@
flushDoneCh chan struct{} // Signals when flush is complete
flushTimer *time.Timer
timerMu sync.Mutex
+ flushMu sync.Mutex
lastAddTime time.TimeAnd at the beginning of Flush():
func (s *S3Connector) Flush() error {
+ s.flushMu.Lock()
+ defer s.flushMu.Unlock()Longer-term: replace flushDoneCh with a per-flush ack (e.g., send a chan struct{} via flushCh) or use a sync.Cond to broadcast completion to all waiters.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| select { | |
| case <-s.flushDoneCh: | |
| return nil | |
| case <-time.After(60 * time.Second): | |
| return fmt.Errorf("flush timeout after 60 seconds") | |
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | |
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | |
| } | |
| default: | |
| // Flush channel is full, likely a flush is already in progress | |
| // Wait for it to complete | |
| select { | |
| case <-s.flushDoneCh: | |
| return nil | |
| case <-time.After(60 * time.Second): | |
| return fmt.Errorf("flush timeout after 60 seconds") | |
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | |
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | |
| } | |
| } | |
| type S3Connector struct { | |
| // ... | |
| flushDoneCh chan struct{} // Signals when flush is complete | |
| flushTimer *time.Timer | |
| timerMu sync.Mutex | |
| flushMu sync.Mutex | |
| lastAddTime time.Time | |
| } | |
| func (s *S3Connector) Flush() error { | |
| s.flushMu.Lock() | |
| defer s.flushMu.Unlock() | |
| select { | |
| case <-s.flushDoneCh: | |
| return nil | |
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | |
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | |
| } | |
| default: | |
| // Flush channel is full, likely a flush is already in progress | |
| // Wait for it to complete | |
| select { | |
| case <-s.flushDoneCh: | |
| return nil | |
| case <-time.After(time.Duration(s.config.FlushTimeout) * time.Second): | |
| return fmt.Errorf("flush timeout after %d seconds", s.config.FlushTimeout) | |
| } | |
| } | |
| } |
jakeloo
approved these changes
Sep 11, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Summary by CodeRabbit
New Features
Bug Fixes