setup workmode determination and update #224
Conversation
|
Caution Review failedThe pull request is closed. WalkthroughA new WorkModeMonitor component was introduced to manage "live" and "backfill" operational modes based on blockchain sync status. The Poller and Committer components now support dynamic work mode switching via channels. Metrics were added to track the current mode. The orchestrator was updated to coordinate these changes, enabling runtime mode transitions. Changes
Sequence Diagram(s)sequenceDiagram
participant Orchestrator
participant WorkModeMonitor
participant Poller
participant Committer
participant Metrics
Orchestrator->>WorkModeMonitor: Start (as goroutine)
Orchestrator->>Poller: Create with workModeChan
Orchestrator->>Committer: Create with workModeChan
WorkModeMonitor->>Poller: Send WorkMode (via channel)
WorkModeMonitor->>Committer: Send WorkMode (via channel)
WorkModeMonitor->>Metrics: Update CurrentWorkMode metric
Poller->>Poller: Shutdown on WorkModeLive
Committer->>Committer: Update mode on channel receive
sequenceDiagram
participant WorkModeMonitor
participant Blockchain (rpc)
participant Storage
loop Every 10 minutes
WorkModeMonitor->>Blockchain: Get latest block number
WorkModeMonitor->>Storage: Get last committed block number
WorkModeMonitor->>WorkModeMonitor: Determine mode (live/backfill)
WorkModeMonitor->>WorkModeMonitor: Broadcast mode if changed
end
📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (6)
✨ Finishing Touches
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
This stack of pull requests is managed by Graphite. Learn more about stacking. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🔭 Outside diff range comments (1)
internal/orchestrator/orchestrator.go (1)
81-91: 🛠️ Refactor suggestionAdd defer statement for channel cleanup to handle early exits.
Similar to the poller, ensure the committer's work mode channel is unregistered even if it exits early.
if o.committerEnabled { wg.Add(1) go func() { defer wg.Done() committerWorkModeChan := make(chan WorkMode, 1) workModeMonitor.RegisterChannel(committerWorkModeChan) + defer workModeMonitor.UnregisterChannel(committerWorkModeChan) committer := NewCommitter(o.rpc, o.storage, WithCommitterWorkModeChan(committerWorkModeChan)) committer.Start(ctx) - workModeMonitor.UnregisterChannel(committerWorkModeChan) }() }
🧹 Nitpick comments (3)
internal/orchestrator/orchestrator.go (1)
57-59: Consider adding error handling for WorkModeMonitor creation.While
NewWorkModeMonitordoesn't return an error currently, consider future-proofing by designing it to return an error if initialization could fail.internal/orchestrator/committer.go (1)
30-31: Consider documenting the workMode field usage.Since
workModeis currently written but never read, consider adding a comment explaining its future use or current purpose for tracking.+ // workMode tracks the current operational mode (backfill or live) workMode WorkMode workModeChan chan WorkModeinternal/orchestrator/work_mode_monitor.go (1)
115-122: Consider implementing reliable message delivery for critical mode changes.The non-blocking send pattern could drop work mode notifications if channels are full. For critical state changes like work mode transitions, consider implementing a retry mechanism or increasing channel buffer sizes.
Consider using a blocking send with timeout:
for ch := range m.workModeChannels { - select { - case ch <- mode: - log.Debug().Msg("Work mode change notification sent") - default: - log.Warn().Msg("Work mode change notification dropped - channel full") - } + go func(ch chan WorkMode) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + select { + case ch <- mode: + log.Debug().Msg("Work mode change notification sent") + case <-ctx.Done(): + log.Warn().Msg("Work mode change notification timed out") + } + }(ch) }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
internal/metrics/metrics.go(1 hunks)internal/orchestrator/committer.go(3 hunks)internal/orchestrator/orchestrator.go(3 hunks)internal/orchestrator/poller.go(4 hunks)internal/orchestrator/work_mode_monitor.go(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
internal/orchestrator/poller.go (4)
internal/orchestrator/work_mode_monitor.go (2)
WorkMode(15-15)WorkModeLive(20-20)internal/rpc/rpc.go (1)
IRPCClient(42-54)internal/storage/connector.go (1)
IStorage(67-71)configs/config.go (1)
Cfg(181-181)
🔇 Additional comments (5)
internal/metrics/metrics.go (1)
148-154: LGTM!The new Prometheus gauge metric is well-defined and follows the established patterns in the codebase.
internal/orchestrator/committer.go (1)
81-85: LGTM!The non-blocking channel receive pattern for work mode updates is well implemented.
internal/orchestrator/work_mode_monitor.go (2)
125-148: LGTM!The work mode determination logic is clear and correct. The 500-block threshold provides a good buffer for determining when the system is caught up.
42-54:⚠️ Potential issuePotential race condition when reading currentMode.
The
currentModefield is read without holding the mutex. While this is likely safe due to the single-writer pattern, it's better to be explicit about synchronization.func (m *WorkModeMonitor) RegisterChannel(ch chan WorkMode) { m.channelsMutex.Lock() defer m.channelsMutex.Unlock() m.workModeChannels[ch] = struct{}{} + currentMode := m.currentMode + m.channelsMutex.Unlock() + // Send current mode to the new channel select { - case ch <- m.currentMode: + case ch <- currentMode: log.Debug().Msg("Initial work mode sent to new channel") default: log.Warn().Msg("Failed to send initial work mode to new channel - channel full") } + return }Actually, on second thought, since we need to hold the lock while accessing the channels map, let's keep the current approach but add a comment:
func (m *WorkModeMonitor) RegisterChannel(ch chan WorkMode) { m.channelsMutex.Lock() defer m.channelsMutex.Unlock() m.workModeChannels[ch] = struct{}{} + // Note: currentMode is read here while holding channelsMutex, which provides + // sufficient synchronization since mode changes also acquire this lock via broadcastWorkMode // Send current mode to the new channel select { case ch <- m.currentMode:Likely an incorrect or invalid review comment.
internal/orchestrator/poller.go (1)
296-301: LGTM!The shutdown method properly consolidates the cleanup logic with correct ordering: cancel context, close channel, and wait for workers.
iuwqyir
force-pushed
the
06-11-setup_workmode_determination_and_update
branch
from
6a68db3 to
5628b1f
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
internal/orchestrator/work_mode_monitor.go (2)
143-145: Avoid potential overflow in debug log
blockDiff.Int64()will truncate/overflow once the difference exceeds
math.MaxInt64. Use.String()to log the exact value:- log.Debug().Msgf("Committer is %d blocks behind the chain", blockDiff.Int64()) + log.Debug().Msgf("Committer is %s blocks behind the chain", blockDiff.String())
17-22: Follow Go naming conventions for constantsALL_CAPS is reserved for C; Go prefers
CamelCase. Renaming promotes
consistency across the codebase:- const ( - WORK_MODE_CHECK_INTERVAL = 10 * time.Minute - WORK_MODE_BACKFILL_THRESHOLD = 500 + const ( + workModeCheckInterval = 10 * time.Minute + workModeBackfillThreshold = 500(remember to update call-sites).
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
internal/metrics/metrics.go(1 hunks)internal/orchestrator/committer.go(3 hunks)internal/orchestrator/orchestrator.go(3 hunks)internal/orchestrator/poller.go(4 hunks)internal/orchestrator/work_mode_monitor.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (4)
- internal/metrics/metrics.go
- internal/orchestrator/orchestrator.go
- internal/orchestrator/committer.go
- internal/orchestrator/poller.go
🧰 Additional context used
🪛 GitHub Actions: Go Code Format Check
internal/orchestrator/work_mode_monitor.go
[error] 1-1: gofmt formatting check failed. The file needs to be formatted.
🔇 Additional comments (1)
internal/orchestrator/work_mode_monitor.go (1)
1-151:gofmtfailure blocks CIThe Go formatting job is red. Run
gofmt -s -w internal/orchestrator/work_mode_monitor.goand commit the result.
🧰 Tools
🪛 GitHub Actions: Go Code Format Check
[error] 1-1: gofmt formatting check failed. The file needs to be formatted.
iuwqyir
force-pushed
the
06-11-setup_workmode_determination_and_update
branch
2 times, most recently
from
5305187 to
6e15133
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
internal/orchestrator/committer_test.go (1)
407-410: Repeat of direct field mutation – see previous commentSame concern as above: direct mutation of
committer.workModecouples the test to internals and may skip required initialization logic. Apply the suggested constructor/option refactor here as well.
🧹 Nitpick comments (1)
internal/orchestrator/committer_test.go (1)
373-376: Expose an explicit constructor/option for initial work-mode instead of mutating a private fieldThe tests reach inside the
committerand mutateworkModedirectly.
This:
- Couples the tests to an internal implementation detail, making refactors harder.
- Bypasses any invariants or side-effects that may be performed when the committer’s mode is changed (e.g. publishing the initial value on
workModeChan, metrics updates, etc.).Prefer providing a public functional option such as
WithInitialWorkMode(mode WorkMode)(mirroring the pattern already introduced for other parameters) or a setter exposed specifically for tests.
Example sketch:committer := NewCommitter(mockRPC, mockStorage, -) -committer.workMode = WorkModeBackfill + WithInitialWorkMode(WorkModeBackfill), +)This keeps the tests resilient to future internal changes and guarantees that mode-related wiring is executed consistently.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
internal/metrics/metrics.go(1 hunks)internal/orchestrator/committer.go(3 hunks)internal/orchestrator/committer_test.go(2 hunks)internal/orchestrator/orchestrator.go(3 hunks)internal/orchestrator/poller.go(4 hunks)internal/orchestrator/work_mode_monitor.go(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (5)
- internal/metrics/metrics.go
- internal/orchestrator/orchestrator.go
- internal/orchestrator/committer.go
- internal/orchestrator/work_mode_monitor.go
- internal/orchestrator/poller.go
🧰 Additional context used
🧬 Code Graph Analysis (1)
internal/orchestrator/committer_test.go (1)
internal/orchestrator/work_mode_monitor.go (1)
WorkModeBackfill(21-21)
iuwqyir
force-pushed
the
06-11-setup_workmode_determination_and_update
branch
from
6e15133 to
3fb9870
Compare
TL;DR
Added a work mode monitor to dynamically switch between backfill and live processing modes.
What changed?
WorkModeMonitorcomponent that determines whether the system should operate in "backfill" or "live" mode based on how far behind the indexer is from the chain headHow to test?
current_work_mode(0 = backfill, 1 = live)Why make this change?
This change improves resource utilization by dynamically adapting the system's behavior based on its current state:
Summary by CodeRabbit
New Features
Bug Fixes