re2: Add attempt metrics in dev (prod WIP). Added max concurrent runs setting to dev using p-limit #1766
Conversation
…ing to dev using p-limit
|
WalkthroughThis pull request enhances the development environment's configuration and execution monitoring. It introduces a new property for maximum concurrent runs in the environment schema and propagates this setting into API responses and CLI commands. The changes integrate a concurrency limiter via the Changes
Sequence Diagram(s)sequenceDiagram
participant CLI as Dev Command CLI
participant Supervisor as DevSupervisor
participant Limiter as p-limit
participant Controller as DevRunController
participant Worker as Run Worker
CLI->>Supervisor: Invoke dev run (with --max-concurrent-runs)
Supervisor->>Limiter: Initialize limiter with maxConcurrentRuns
Supervisor->>Limiter: Schedule run execution
Limiter->>Controller: Execute run (passing dequeuedAt & metrics)
Controller->>Worker: Start and execute run attempt
Worker-->>Controller: Return result and execution metrics
Controller-->>Limiter: Signal completion
Limiter-->>Supervisor: Release concurrency slot
Possibly related PRs
Suggested reviewers
Poem
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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 (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (6)
apps/webapp/app/env.server.ts (1)
574-576: Consider adding a minimum concurrency constraint.While the default of 25 is sensible, you might want to safeguard against invalid or negative inputs (e.g.,
0or negative values) by applying an explicit minimum of 1 to ensurez.coerce.number()never evaluates to a concurrency thatpLimitrejects.-DEV_MAX_CONCURRENT_RUNS: z.coerce.number().int().default(25), +DEV_MAX_CONCURRENT_RUNS: z.coerce.number().int().min(1).default(25),packages/cli-v3/src/commands/dev.ts (2)
22-22: Suggest providing a default or validation for maxConcurrentRuns.Currently, the optional property without a default or validation could allow unintended zero or negative concurrency if specified incorrectly. Consider enforcing a minimum or reusing the environment default for safety.
- maxConcurrentRuns: z.coerce.number().optional(), + maxConcurrentRuns: z.coerce.number().min(1).optional(),
41-44: Optional short CLI alias.Having a short alias (e.g.,
-m) for--max-concurrent-runscan improve developer ergonomics. Also ensure the help text clarifies that it's merged with, but cannot exceed, the environment limit (by default 25).packages/cli-v3/src/dev/devSupervisor.ts (2)
87-91: Validate negative or zero concurrency edge cases.While
Math.minensures we don’t exceed the environment’s concurrency, passing0or a negative number can cause runtime issues. Consider clamping the user-supplied or config concurrency to a minimum of 1.-const maxConcurrentRuns = Math.min( - this.config.maxConcurrentRuns, - this.options.args.maxConcurrentRuns ?? this.config.maxConcurrentRuns -); +const rawCliRuns = this.options.args.maxConcurrentRuns ?? this.config.maxConcurrentRuns; +const maxConcurrentRuns = Math.max(1, Math.min(this.config.maxConcurrentRuns, rawCliRuns));
193-200: Use >= instead of > for concurrency checks.If the concurrency limit is 5, you might want to defer more dequeues once the sum of active + pending hits 5 (rather than 6).
-if ( - this.runLimiter && - this.runLimiter.activeCount + this.runLimiter.pendingCount > this.runLimiter.concurrency -) { +if ( + this.runLimiter && + this.runLimiter.activeCount + this.runLimiter.pendingCount >= this.runLimiter.concurrency +) {packages/cli-v3/src/entryPoints/managed-run-worker.ts (1)
267-282: Task import now instrumented with metrics.The task import process is now wrapped in a metrics measurement block to track import time, following a structured approach to performance monitoring.
One small observation: both the metrics measurement and manual timing calculation are performed. While this might be intentional for logging purposes, consider eventually consolidating to use just the metrics system for timing in a future PR.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (12)
apps/webapp/app/env.server.ts(1 hunks)apps/webapp/app/routes/engine.v1.dev.config.ts(1 hunks)internal-packages/run-engine/src/engine/index.ts(1 hunks)packages/cli-v3/package.json(1 hunks)packages/cli-v3/src/commands/dev.ts(2 hunks)packages/cli-v3/src/dev/devSupervisor.ts(5 hunks)packages/cli-v3/src/entryPoints/dev-run-controller.ts(7 hunks)packages/cli-v3/src/entryPoints/dev-run-worker.ts(5 hunks)packages/cli-v3/src/entryPoints/managed-run-worker.ts(5 hunks)packages/core/src/v3/schemas/api.ts(1 hunks)packages/core/src/v3/schemas/runEngine.ts(1 hunks)references/v3-catalog/package.json(2 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (7)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: units / 🧪 Unit Tests
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (25)
packages/cli-v3/package.json (1)
113-113: Added p-limit dependency for concurrency controlThe addition of p-limit (v6.2.0) is appropriate for implementing the maximum concurrent runs feature mentioned in the PR title. This library provides a clean way to limit the number of promises running simultaneously.
references/v3-catalog/package.json (1)
9-9: Script name simplification looks goodRenaming from "dev:trigger" to "dev" while keeping the same command makes the scripts more consistent and easier to use.
apps/webapp/app/routes/engine.v1.dev.config.ts (1)
23-23:✅ Verification successful
Added maxConcurrentRuns to the dev config API response
This change properly exposes the maximum concurrent runs setting to the API response, which aligns with the PR objective of adding this setting to the development environment.
🏁 Script executed:
#!/bin/bash # Verify that the DevConfigResponseBody schema includes the maxConcurrentRuns property ast-grep --pattern 'export const DevConfigResponseBody = z.object({ $$$ maxConcurrentRuns: $$$, $$$ })'Length of output: 544
Changes Verified: maxConcurrentRuns is correctly exposed in both the dev config and its API schema.
The addition at
apps/webapp/app/routes/engine.v1.dev.config.ts(line 23) now properly maps themaxConcurrentRunsvalue (sourced fromenv.DEV_MAX_CONCURRENT_RUNS) to the API response. Verification confirms that theDevConfigResponseBodyschema inpackages/core/src/v3/schemas/api.tsincludes this property as a number.No further changes are needed.
packages/core/src/v3/schemas/runEngine.ts (1)
142-142:❓ Verification inconclusive
Added dequeuedAt timestamp for better metrics tracking
Adding a timestamp for when messages are dequeued is valuable for monitoring and calculating queue processing metrics. This change uses the appropriate z.coerce.date() to ensure proper type conversion.
🏁 Script executed:
#!/bin/bash # Check if the dequeuedAt field is properly set when dequeuing messages ast-grep --pattern 'dequeueFromMasterQueue($$$) { $$$ dequeuedAt: $$$, $$$ }'Length of output: 85
Action Required: Confirm integration of the 'dequeuedAt' field with the dequeuing logic
The change in
packages/core/src/v3/schemas/runEngine.tscorrectly adds thedequeuedAt: z.coerce.date(),entry to support metrics tracking by ensuring proper date conversion. However, our initial pattern search in a function context (i.e.dequeueFromMasterQueue) did not return any matching results. Please manually verify that this timestamp is effectively utilized within the proper message dequeueing process and that its integration aligns with your intended processing logic.internal-packages/run-engine/src/engine/index.ts (1)
917-917: Added timestamp for dequeued messagesThis change adds a
dequeuedAttimestamp property to the dequeued message object, which will help track when a run was dequeued from the master queue. This timing information can be used to calculate queue wait time and other performance metrics.packages/core/src/v3/schemas/api.ts (1)
430-430: Added maximum concurrent runs configuration to DevConfigResponseBodyThis property will support the PR objective of limiting concurrent runs in the development environment using p-limit. The schema now properly defines the expected type as a number.
packages/cli-v3/src/dev/devSupervisor.ts (3)
27-27: Ensure p-limit is properly declared as a direct dependency.If
p-limitis a newly introduced package, verify it exists in your dependency list to avoid runtime errors.
69-70: Good approach to store concurrency limiter in a class field.Storing the limiter instance at the class level is a clean design for consistent concurrency management across methods.
92-94: Handle invalid concurrency constants when calling pLimit.
pLimitwill throw if concurrency is less than 1. ConfirmmaxConcurrentRunsis at least 1 to prevent unhandled exceptions.packages/cli-v3/src/entryPoints/managed-run-worker.ts (4)
20-20: Import added for metrics tracking capability.The addition of
runTimelineMetricsis consistent with the PR objective of adding metrics measurement capabilities.
41-41: Import added for metrics management implementation.The
StandardRunTimelineMetricsManagerclass provides the concrete implementation for tracking execution timeline metrics.
96-98: Setup global metrics manager for tracking execution timelines.This initialization block establishes the metrics tracking infrastructure, following the same pattern used for other global managers in the codebase.
208-210: Handler updated to receive and register metrics.The
EXECUTE_TASK_RUNhandler has been updated to receive metrics from the execution context and register them with the metrics manager. This ensures metrics from the controller are properly integrated.packages/cli-v3/src/entryPoints/dev-run-worker.ts (6)
20-20: Import added for metrics tracking capability.The addition of
runTimelineMetricsis consistent with the PR objective of adding metrics measurement capabilities.
40-40: Import added for metrics management implementation.The
StandardRunTimelineMetricsManagerclass provides the concrete implementation for tracking execution timeline metrics.
92-94: Setup global metrics manager for tracking execution timelines.This initialization block establishes the metrics tracking infrastructure, following the same pattern used for other global managers in the codebase.
198-198: Handler signature updated to include metrics parameter.The method signature update ensures metrics can be passed from the controller.
201-201: Register metrics from execution context.This line ensures that metrics passed from the controller are properly registered with the metrics manager.
257-272: Task import now instrumented with metrics.The task import process is now wrapped in a metrics measurement block to track import time, providing consistent metrics tracking between the dev and managed workers.
packages/cli-v3/src/entryPoints/dev-run-controller.ts (6)
8-8: Add import for metrics type definition.The
TaskRunExecutionMetricstype is imported to support the new metrics tracking functionality.
479-485: Added dequeue timestamp parameter to capture queue time.The
dequeuedAtparameter allows tracking when a run attempt was dequeued from the queue, which is essential for calculating queue wait times.
492-504: Added timing measurement for attempt execution.This code block captures the start time and calculates the duration of the attempt, providing valuable metrics about the execution process.
518-537: Construct metrics array with execution timeline events.The metrics array captures key events in the execution timeline, including attempt creation and dequeue times. The conditional inclusion of the dequeue event is a good approach to handle cases where the timestamp might not be available.
596-599: Added metrics parameter to executeRun method.The method signature update ensures that the metrics can be passed to the worker for consistent tracking across the execution pipeline.
787-787: Passing dequeue timestamp from controller to execution context.This line ensures that the dequeue timestamp is passed from the controller to the worker, enabling accurate measurement of the time spent in the queue.
| if (this.runLimiter) { | ||
| this.runLimiter(() => runController.start(message)).then(() => { | ||
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | ||
| }); | ||
| } else { | ||
| //don't await for run completion, we want to dequeue more runs | ||
| runController.start(message).then(() => { | ||
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | ||
| }); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider adding error handling for run start.
Currently, if runController.start(message) rejects, no error path is handled. Logging or capturing that error will help diagnose startup failures.
-this.runLimiter(() => runController.start(message)).then(() => {
+this.runLimiter(() => runController.start(message))
+ .then(() => {
logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId });
})
+ .catch((error) => {
+ logger.debug("[DevSupervisor] Run failed to start", {
+ runId: message.run.friendlyId,
+ error
+ });
+ });📝 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.
| if (this.runLimiter) { | |
| this.runLimiter(() => runController.start(message)).then(() => { | |
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | |
| }); | |
| } else { | |
| //don't await for run completion, we want to dequeue more runs | |
| runController.start(message).then(() => { | |
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | |
| }); | |
| } | |
| if (this.runLimiter) { | |
| this.runLimiter(() => runController.start(message)) | |
| .then(() => { | |
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | |
| }) | |
| .catch((error) => { | |
| logger.debug("[DevSupervisor] Run failed to start", { | |
| runId: message.run.friendlyId, | |
| error | |
| }); | |
| }); | |
| } else { | |
| //don't await for run completion, we want to dequeue more runs | |
| runController.start(message).then(() => { | |
| logger.debug("[DevSupervisor] Run started", { runId: message.run.friendlyId }); | |
| }); | |
| } |
Summary by CodeRabbit
New Features
Chores