v4: current concurrency sweeper #2206
Conversation
|
WalkthroughThis change introduces a concurrency sweeper feature to the run engine and run queue subsystems. New environment variables are added for configuring the sweeper's scan interval, processing interval, and log levels. The ✨ 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:
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 (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (6)
internal-packages/run-engine/src/run-queue/tests/concurrencySweeper.test.ts (2)
52-52: Consider reducing the test timeout.A 60-second timeout seems excessive for this test. Consider reducing it to a more reasonable value (e.g., 10-15 seconds) to prevent long test runs while still allowing adequate time for Redis operations.
-vi.setConfig({ testTimeout: 60_000 }); +vi.setConfig({ testTimeout: 15_000 });
154-154: Consider making the delay more deterministic.The fixed 1-second delay might be brittle in CI environments. Consider using a polling approach or reducing the sweeper intervals in the test to make it more deterministic.
- await setTimeout(1000); // Now a run is "completed" and should be removed from the concurrency set + // Poll for the expected result with shorter intervals + let attempts = 0; + const maxAttempts = 10; + while (attempts < maxAttempts) { + await setTimeout(100); + const currentConcurrency = await queue.currentConcurrencyOfQueue( + authenticatedEnvDev, + messageDev.queue + ); + if (currentConcurrency === 1) break; + attempts++; + }internal-packages/run-engine/src/run-queue/keyProducer.ts (2)
226-228: Consider extracting the fixed string as a constant.The method returns a hardcoded string "markedForAck". Consider adding this to the
constantsobject at the top of the file for consistency with other key parts.const constants = { CURRENT_CONCURRENCY_PART: "currentConcurrency", CONCURRENCY_LIMIT_PART: "concurrency", DISABLED_CONCURRENCY_LIMIT_PART: "disabledConcurrency", ENV_PART: "env", ORG_PART: "org", PROJECT_PART: "proj", QUEUE_PART: "queue", CONCURRENCY_KEY_PART: "ck", TASK_PART: "task", MESSAGE_PART: "message", DEAD_LETTER_QUEUE_PART: "deadLetter", MASTER_QUEUE_PART: "masterQueue", WORKER_QUEUE_PART: "workerQueue", + MARKED_FOR_ACK: "markedForAck", } as const;Then update the method:
markedForAckKey(): string { - return "markedForAck"; + return constants.MARKED_FOR_ACK; }
230-232: Consider the performance impact of the leading wildcard in the scan pattern.The pattern
*:${constants.ENV_PART}:*:queue:*:${constants.CURRENT_CONCURRENCY_PART}starts with a wildcard, which can be expensive in Redis SCAN operations, especially with large key spaces. Consider if a more specific pattern prefix could be used to improve scan performance.internal-packages/run-engine/src/engine/index.ts (1)
1345-1383: Consider reducing logging verbosity for production use.The callback logs each found finished run at info level (line 1370), which could generate excessive logs in production environments with high throughput. Consider:
- Changing the log level to
debug- Logging only a summary (e.g., "Found X finished runs")
- Making the 10-minute delay configurable
- // Log the finished runs - for (const run of runs) { - this.logger.info("Concurrency sweeper callback found finished run", { - runId: run.id, - orgId: run.organizationId, - status: run.status, - }); - } + // Log summary instead of individual runs + if (runs.length > 0) { + this.logger.debug("Concurrency sweeper callback found finished runs", { + count: runs.length, + runIds: runs.map(r => r.id).slice(0, 5), // Log first 5 for debugging + }); + }internal-packages/run-engine/src/run-queue/index.ts (1)
2176-2223: Consider adding initial delay and reducing startup load.The sweeper starts both loops immediately with
setTimeout(..., 0). Consider adding a small initial delay to avoid immediate load on startup, especially when multiple instances start simultaneously.start() { this.logger.info("Starting concurrency sweeper"); // Start the scanning process - this._scanInterval = setTimeout(() => { + this._scanInterval = setTimeout(() => { const scanLoop = () => { // ... existing code }; scanLoop(); - }, 0); + }, 5000); // 5 second initial delay // Start the marked runs processing - this._processInterval = setTimeout(() => { + this._processInterval = setTimeout(() => { const processLoop = () => { // ... existing code }; processLoop(); - }, 0); + }, 2500); // 2.5 second initial delay, staggered from scan }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
references/hello-world/trigger.config.tsis excluded by!references/**
📒 Files selected for processing (9)
apps/webapp/app/env.server.ts(2 hunks)apps/webapp/app/v3/runEngine.server.ts(4 hunks)internal-packages/run-engine/src/engine/index.ts(7 hunks)internal-packages/run-engine/src/engine/statuses.ts(1 hunks)internal-packages/run-engine/src/engine/types.ts(3 hunks)internal-packages/run-engine/src/run-queue/index.ts(10 hunks)internal-packages/run-engine/src/run-queue/keyProducer.ts(1 hunks)internal-packages/run-engine/src/run-queue/tests/concurrencySweeper.test.ts(1 hunks)internal-packages/run-engine/src/run-queue/types.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (3)
`apps/webapp/app/v3/runEngine.server.ts`: This file configures the run engine.
apps/webapp/app/v3/runEngine.server.ts: This file configures the run engine.
📄 Source: CodeRabbit Inference Engine (.cursor/rules/webapp.mdc)
List of files the instruction was applied to:
apps/webapp/app/v3/runEngine.server.ts
`apps/webapp/**/*`: The webapp at apps/webapp is a Remix 2.1 app that uses Node.js v20. Use zod extensively in the webapp.
apps/webapp/**/*: The webapp at apps/webapp is a Remix 2.1 app that uses Node.js v20.
Use zod extensively in the webapp.
📄 Source: CodeRabbit Inference Engine (.github/copilot-instructions.md)
List of files the instruction was applied to:
apps/webapp/app/v3/runEngine.server.tsapps/webapp/app/env.server.ts
`apps/webapp/app/env.server.ts`: All environment variables in the webapp must be accessed through the 'env' export of env.server.ts, never directly via process.env.
apps/webapp/app/env.server.ts: All environment variables in the webapp must be accessed through the 'env' export of env.server.ts, never directly via process.env.
📄 Source: CodeRabbit Inference Engine (.cursor/rules/webapp.mdc)
List of files the instruction was applied to:
apps/webapp/app/env.server.ts
🧠 Learnings (8)
📓 Common learnings
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
internal-packages/run-engine/src/engine/statuses.ts (1)
Learnt from: nicktrn
PR: triggerdotdev/trigger.dev#1418
File: packages/core/src/v3/errors.ts:364-371
Timestamp: 2024-10-18T15:41:52.352Z
Learning: In `packages/core/src/v3/errors.ts`, within the `taskRunErrorEnhancer` function, `error.message` is always defined, so it's safe to directly call `error.message.includes("SIGTERM")` without additional checks.
internal-packages/run-engine/src/run-queue/tests/concurrencySweeper.test.ts (1)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
apps/webapp/app/v3/runEngine.server.ts (6)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: When writing tests in the webapp (apps/webapp/app/**/*.test.{ts,tsx}), avoid importing env.server.ts directly or indirectly; instead, ensure that all dependencies are testable and receive configuration via options, not environment variables.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The Prisma client (from @trigger.dev/database) is set up in db.server.ts and should be used for all PostgreSQL access in the webapp.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: Maintain a clear separation between service logic and configuration: implement testable services in dedicated files (e.g., services/realtimeClient.server.ts) and handle configuration in separate files (e.g., services/realtimeClientGlobal.server.ts).
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: In the Remix 2.1.0 webapp (apps/webapp), all environment variables must be accessed via the 'env' export from env.server.ts, never directly from process.env, to ensure consistent and secure environment management.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/executing-commands.mdc:0-0
Timestamp: 2025-06-24T08:06:20.782Z
Learning: Most commands in the monorepo should be executed using 'pnpm run ...' from the root directory, but for running tests, it is often preferable to cd into the specific package or app directory and run the test command locally to enable targeting individual test files.
internal-packages/run-engine/src/engine/types.ts (8)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The Prisma client (from @trigger.dev/database) is set up in db.server.ts and should be used for all PostgreSQL access in the webapp.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: When implementing Trigger.dev tasks in files under '**/trigger/**/*.{ts,tsx,js,jsx}', always import from '@trigger.dev/sdk/v3' and export every task, including subtasks, to ensure correct discovery and execution.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: When triggering tasks from backend code, use 'tasks.trigger', 'tasks.batchTrigger', or 'tasks.triggerAndPoll' from '@trigger.dev/sdk/v3', and use type-only imports for type safety.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: Incorrect implementation of Trigger.dev tasks (wrong imports, missing exports, deprecated patterns) will break production, fail deployment, and fail to run locally.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: The deprecated 'client.defineJob' pattern must never be used in Trigger.dev tasks; always use the 'task' function from '@trigger.dev/sdk/v3' for defining tasks.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: Before generating any Trigger.dev task code, always verify: (1) imports are from '@trigger.dev/sdk/v3', (2) every task is exported, and (3) no deprecated code patterns are present.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: The 'trigger.config.ts' file configures the Trigger.dev project, specifying task directories, retry settings, telemetry, runtime, machine settings, log level, max duration, and build extensions.
internal-packages/run-engine/src/engine/index.ts (4)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The Prisma client (from @trigger.dev/database) is set up in db.server.ts and should be used for all PostgreSQL access in the webapp.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: For real-time updates on task runs, use the 'runs.subscribeToRun', 'runs.subscribeToRunsWithTag', or 'runs.subscribeToBatch' APIs from '@trigger.dev/sdk/v3'.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: The 'run' function in a Trigger.dev task is the main entry point for task logic and can be asynchronous; it receives the payload and context.
apps/webapp/app/env.server.ts (3)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: When writing tests in the webapp (apps/webapp/app/**/*.test.{ts,tsx}), avoid importing env.server.ts directly or indirectly; instead, ensure that all dependencies are testable and receive configuration via options, not environment variables.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: In the Remix 2.1.0 webapp (apps/webapp), all environment variables must be accessed via the 'env' export from env.server.ts, never directly from process.env, to ensure consistent and secure environment management.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
internal-packages/run-engine/src/run-queue/index.ts (3)
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-06-24T08:06:34.164Z
Learning: The legacy run engine code is being deprecated in favor of Run Engine 2.0, which is encapsulated in the @internal/run-engine package; new development should target Run Engine 2.0.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .github/copilot-instructions.md:0-0
Timestamp: 2025-06-24T08:06:17.013Z
Learning: Always prefer using isomorphic code (such as fetch, ReadableStream, etc.) instead of Node.js specific code in TypeScript projects, especially for SDKs intended to run in multiple environments.
Learnt from: CR
PR: triggerdotdev/trigger.dev#0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-06-24T08:07:10.028Z
Learning: For real-time updates on task runs, use the 'runs.subscribeToRun', 'runs.subscribeToRunsWithTag', or 'runs.subscribeToBatch' APIs from '@trigger.dev/sdk/v3'.
🧬 Code Graph Analysis (2)
internal-packages/run-engine/src/engine/statuses.ts (1)
packages/core/src/v3/schemas/runEngine.ts (2)
TaskRunStatus(21-38)TaskRunStatus(40-40)
internal-packages/run-engine/src/engine/types.ts (2)
packages/core/src/logger.ts (1)
LogLevel(15-15)internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.ts (1)
FairQueueSelectionStrategyOptions(36-50)
⏰ Context from checks skipped due to timeout of 90000ms (25)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (10, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (9, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 10)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 10)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 10)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- 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 - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: typecheck / typecheck
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (11)
internal-packages/run-engine/src/engine/statuses.ts (2)
44-53: LGTM: Clean refactoring for better reusability.Moving the
finalStatusesarray to module level is a good design decision that enables reuse across multiple functions and provides a cleaner API for the concurrency sweeper feature.
59-61: LGTM: Well-designed export function.The
getFinalRunStatuses()function provides a clean way for external modules to access the final statuses array, which is essential for the concurrency sweeper functionality.internal-packages/run-engine/src/run-queue/types.ts (1)
100-102: LGTM: Well-designed interface extensions.The new concurrency sweeper methods follow consistent naming conventions and provide appropriate abstractions for Redis key management. The clear comment section helps organize the interface methods by functionality.
internal-packages/run-engine/src/run-queue/tests/concurrencySweeper.test.ts (1)
75-87: LGTM: Well-structured concurrency sweeper configuration.The test properly configures the concurrency sweeper with appropriate intervals and a realistic callback that simulates identifying completed runs. The conditional logic based on
enableConcurrencySweeperallows for controlled testing of the feature.apps/webapp/app/v3/runEngine.server.ts (3)
2-2: LGTM: Proper import of read-only replica.Adding the
$replicaimport enables the use of a read-only database client for the concurrency sweeper, which is a good practice to avoid impacting the primary database with read operations.
16-16: LGTM: Read-only database client configuration.Using the read-only replica for the concurrency sweeper is an excellent design decision that will help distribute database load and avoid impacting write operations on the primary database.
43-43: LGTM: Well-structured configuration for concurrency sweeper.The configuration properly integrates the new environment variables for logging and concurrency sweeper timing. The structure follows existing patterns and provides appropriate controls for the feature.
Also applies to: 69-73
apps/webapp/app/env.server.ts (2)
431-435: LGTM: Well-designed environment variables with sensible defaults.The new concurrency sweeper configuration variables follow established naming conventions and provide reasonable default values. The scan interval (60s) being longer than the processing interval (5s) creates a good balance between performance and responsiveness.
601-601: LGTM: Consistent log level configuration.The run queue log level configuration follows the same pattern as other log level environment variables in the codebase, maintaining consistency across the application.
internal-packages/run-engine/src/engine/types.ts (1)
11-20: LGTM! Good use of read-only replica for the sweeper.The addition of the optional
readOnlyPrismaproperty is a good architectural decision, allowing the concurrency sweeper to use a read replica and reduce load on the primary database.internal-packages/run-engine/src/engine/index.ts (1)
140-147: Good implementation of concurrency sweeper configuration.The concurrency sweeper is properly integrated with sensible defaults (60s scan interval, 5s process interval) and correctly disabled when the worker is disabled.
| */ | ||
| export class RunQueue { | ||
| private retryOptions: RetryOptions; | ||
| private subscriber: Redis; | ||
| private luaDebugSubscriber: Redis; | ||
| private logger: Logger; | ||
| private redis: Redis; | ||
| public redis: Redis; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider keeping redis property private for better encapsulation.
Making the redis property public breaks encapsulation. Consider providing specific methods that the ConcurrencySweeper needs instead of exposing the entire Redis client.
🤖 Prompt for AI Agents
In internal-packages/run-engine/src/run-queue/index.ts at line 144, the redis
property is currently public, which breaks encapsulation. Change the redis
property to private and create specific public methods that expose only the
necessary Redis functionality required by the ConcurrencySweeper, instead of
exposing the entire Redis client directly.
| async processMarkedRun(run: MarkedRun) { | ||
| this.logger.info("Acknowledging marked run", { | ||
| orgId: run.orgId, | ||
| messageId: run.messageId, | ||
| }); | ||
|
|
||
| await this.runQueue.acknowledgeMessage(run.orgId, run.messageId, { | ||
| skipDequeueProcessing: true, | ||
| removeFromWorkerQueue: false, | ||
| }); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use debug level for per-run acknowledgment logging.
Similar to the previous comment, logging each individual acknowledgment at info level will generate excessive logs.
async processMarkedRun(run: MarkedRun) {
- this.logger.info("Acknowledging marked run", {
+ this.logger.debug("Acknowledging marked run", {
orgId: run.orgId,
messageId: run.messageId,
});
await this.runQueue.acknowledgeMessage(run.orgId, run.messageId, {
skipDequeueProcessing: true,
removeFromWorkerQueue: false,
});
}📝 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.
| async processMarkedRun(run: MarkedRun) { | |
| this.logger.info("Acknowledging marked run", { | |
| orgId: run.orgId, | |
| messageId: run.messageId, | |
| }); | |
| await this.runQueue.acknowledgeMessage(run.orgId, run.messageId, { | |
| skipDequeueProcessing: true, | |
| removeFromWorkerQueue: false, | |
| }); | |
| } | |
| async processMarkedRun(run: MarkedRun) { | |
| - this.logger.info("Acknowledging marked run", { | |
| + this.logger.debug("Acknowledging marked run", { | |
| orgId: run.orgId, | |
| messageId: run.messageId, | |
| }); | |
| await this.runQueue.acknowledgeMessage(run.orgId, run.messageId, { | |
| skipDequeueProcessing: true, | |
| removeFromWorkerQueue: false, | |
| }); | |
| } |
🤖 Prompt for AI Agents
In internal-packages/run-engine/src/run-queue/index.ts around lines 2457 to
2467, change the logging level for acknowledging each marked run from info to
debug to reduce excessive log output. Update the this.logger.info call to
this.logger.debug while keeping the log message and metadata the same.
| private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) { | ||
| const markedForAckKey = this.keys.markedForAckKey(); | ||
|
|
||
| // Prepare arguments: alternating orgId, messageId pairs | ||
| const args: string[] = []; | ||
| for (const run of completedRuns) { | ||
| this.logger.info("Marking run for acknowledgment", { | ||
| orgId: run.orgId, | ||
| runId: run.id, | ||
| }); | ||
|
|
||
| args.push(run.orgId); | ||
| args.push(run.id); | ||
| } | ||
|
|
||
| const count = await this.redis.markCompletedRunsForAck(markedForAckKey, ...args); | ||
|
|
||
| this.logger.debug(`Marked ${count} runs for acknowledgment`, { | ||
| markedForAckKey, | ||
| count, | ||
| }); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Reduce logging verbosity in markRunsForAck.
Logging each individual run at info level (line 2395) will generate excessive logs in production. Consider using debug level or logging only a summary.
private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) {
const markedForAckKey = this.keys.markedForAckKey();
// Prepare arguments: alternating orgId, messageId pairs
const args: string[] = [];
for (const run of completedRuns) {
- this.logger.info("Marking run for acknowledgment", {
+ this.logger.debug("Marking run for acknowledgment", {
orgId: run.orgId,
runId: run.id,
});
args.push(run.orgId);
args.push(run.id);
}
const count = await this.redis.markCompletedRunsForAck(markedForAckKey, ...args);
- this.logger.debug(`Marked ${count} runs for acknowledgment`, {
+ this.logger.info(`Marked ${count} runs for acknowledgment`, {
markedForAckKey,
count,
});
}📝 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.
| private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) { | |
| const markedForAckKey = this.keys.markedForAckKey(); | |
| // Prepare arguments: alternating orgId, messageId pairs | |
| const args: string[] = []; | |
| for (const run of completedRuns) { | |
| this.logger.info("Marking run for acknowledgment", { | |
| orgId: run.orgId, | |
| runId: run.id, | |
| }); | |
| args.push(run.orgId); | |
| args.push(run.id); | |
| } | |
| const count = await this.redis.markCompletedRunsForAck(markedForAckKey, ...args); | |
| this.logger.debug(`Marked ${count} runs for acknowledgment`, { | |
| markedForAckKey, | |
| count, | |
| }); | |
| } | |
| private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) { | |
| const markedForAckKey = this.keys.markedForAckKey(); | |
| // Prepare arguments: alternating orgId, messageId pairs | |
| const args: string[] = []; | |
| for (const run of completedRuns) { | |
| this.logger.debug("Marking run for acknowledgment", { | |
| orgId: run.orgId, | |
| runId: run.id, | |
| }); | |
| args.push(run.orgId); | |
| args.push(run.id); | |
| } | |
| const count = await this.redis.markCompletedRunsForAck(markedForAckKey, ...args); | |
| this.logger.info(`Marked ${count} runs for acknowledgment`, { | |
| markedForAckKey, | |
| count, | |
| }); | |
| } |
🤖 Prompt for AI Agents
In internal-packages/run-engine/src/run-queue/index.ts around lines 2389 to
2410, the method markRunsForAck logs each run at info level, causing excessive
log volume in production. Change the log level of the per-run message from info
to debug, or alternatively remove individual run logs and instead log a summary
count after processing all runs to reduce verbosity.
| const deserializedMessage = safeJsonParse(rawMessage); | ||
|
|
||
| const message = OutputPayload.safeParse(deserializedMessage); | ||
|
|
||
| if (!message.success) { | ||
| this.logger.error(`[${this.name}] Failed to parse message`, { | ||
| messageKey, | ||
| error: message.error, | ||
| service: this.name, | ||
| deserializedMessage, | ||
| }); | ||
|
|
||
| return; | ||
| return deserializedMessage as OutputPayload; | ||
| } |
There was a problem hiding this comment.
Security concern: Avoid logging raw message data.
Logging the deserializedMessage on line 471 could potentially expose sensitive data in logs. Consider logging only the error details without the raw message content.
if (!message.success) {
this.logger.error(`[${this.name}] Failed to parse message`, {
messageKey,
error: message.error,
service: this.name,
- deserializedMessage,
});
- return deserializedMessage as OutputPayload;
+ return undefined;
}📝 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.
| const deserializedMessage = safeJsonParse(rawMessage); | |
| const message = OutputPayload.safeParse(deserializedMessage); | |
| if (!message.success) { | |
| this.logger.error(`[${this.name}] Failed to parse message`, { | |
| messageKey, | |
| error: message.error, | |
| service: this.name, | |
| deserializedMessage, | |
| }); | |
| return; | |
| return deserializedMessage as OutputPayload; | |
| } | |
| const deserializedMessage = safeJsonParse(rawMessage); | |
| const message = OutputPayload.safeParse(deserializedMessage); | |
| if (!message.success) { | |
| this.logger.error(`[${this.name}] Failed to parse message`, { | |
| messageKey, | |
| error: message.error, | |
| service: this.name, | |
| }); | |
| return undefined; | |
| } |
🤖 Prompt for AI Agents
In internal-packages/run-engine/src/run-queue/index.ts around lines 461 to 474,
avoid logging the raw deserializedMessage to prevent exposing sensitive data.
Modify the logger.error call to exclude deserializedMessage and only log the
error details, messageKey, and service name to maintain useful context without
risking sensitive information leakage.
No description provided.