fix(batch): optimize processing batch trigger v2 #2841
Conversation
|
WalkthroughAdds four environment variables for batch-queue tuning and exposes them in RunEngine. Reworks fair-queue to a master-driven, batch-oriented flow with mandatory worker-queue routing, new batch claim/release Redis Lua commands, external message lifecycle APIs (heartbeatMessage, getMessageData, completeMessage, releaseMessage, failMessage), and WorkerQueueManager always present. DRR scheduler gains masterQueueLimit and a batched deficit-decrement path. BatchQueue implements worker-consumer loops, batched span tracing, shardCount and workerQueueBlockingTimeoutSeconds options, ConcurrencyManager exposes getAvailableCapacity, multiple type updates, and many tests and test helpers added/updated. Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes 🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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. Comment |
PR Review: fix(batch): optimize processing batch trigger v2OverviewThis PR introduces significant performance optimizations to the batch queue processing system by adding:
Code Quality and Best PracticesStrengths:
Suggestions:
let maxClaimCount = 10; // Default batch size capConsider making this configurable via options rather than a hardcoded value.
for (let i = 0; i < processedFromQueue; i++) {
await this.telemetry.trace(
"recordProcessed",
async (span) => {
// ...
await this.scheduler.recordProcessed!(tenantId, queueId);
},
{ kind: SpanKind.INTERNAL }
);
}This calls Potential Bugs or Issues
await this.workerQueueManager!.push(workerQueueId, messageKey);The non-null assertion is used, but this code path is inside
local consumerId = ARGV[2]The
: options.queueId,to: : undefined,Changes the fallback behavior when Performance ConsiderationsPositive:
Considerations:
for (const group of this.groups) {
const groupId = group.extractGroupId(queue);
const key = this.keys.concurrencyKey(group.name, groupId);
const current = await this.redis.scard(key);
const limit = (await group.getLimit(groupId)) || group.defaultLimit;Each iteration makes 2 Redis calls (SCARD + getLimit). For multiple concurrency groups, consider parallelizing these calls or using a Lua script.
const workerQueueId = "worker-queue";All consumers now share a single worker queue. While this simplifies distribution, it could become a contention point under very high load. Consider documenting this trade-off. Security ConcernsNo significant security concerns identified. The changes are internal queue processing optimizations that don't introduce new attack surfaces. Test CoverageGaps identified:
Recommendations:
SummaryThis is a solid performance optimization PR that introduces batch claiming and shared worker queues. The main areas to address:
🤖 Generated with Claude Code |
|
@claude review this PR please |
|
Claude finished @ericallam's task in 1m 29s —— View job PR Review: fix(batch): optimize processing batch trigger v2
OverviewThis PR introduces significant performance optimizations to the batch queue processing system:
Strengths
Issues & Suggestions1. Unused
|
| Priority | Item |
|---|---|
| 🟡 Recommended | Add tests for new claimBatch and getAvailableCapacity methods |
| 🟡 Recommended | Make batch size cap (10) configurable |
| 🟢 Minor | Remove unused consumerId from Lua script |
| 🟢 Consider | Batch variant for recordProcessed to reduce Redis calls |
|
Review CompleteYour review story is ready! Comment !reviewfast on this PR to re-generate the story. |
![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
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/redis-worker/src/fair-queue/index.ts (1)
759-813: Fix abort signal handling inconsistency in master queue consumer loopThe master queue consumer loop has inconsistent abort error handling compared to the rest of the codebase:
Incorrect error detection: Line 804 checks
error.message === "AbortError"while the worker queue consumer (line 1083), direct consumer loop (line 1160), and reclaim loop (line 1681) all checkerror.name === "AbortError". Thenameproperty is the standard way to identify error types in JavaScript.Non-standard error creation: Line 796 creates a plain
new Error("AbortError")instead of relying on the standard AbortError thrown by async abort signals. This only works because it's checked against the message property, making it fragile and inconsistent.Inconsistent patterns: The direct consumer loop (line 1131) uses
for await (const _ of setInterval(..., { signal: this.abortController.signal })), which properly integrates with the abort signal and throws a standard AbortError. The master queue loop should follow the same pattern or at minimum align its error checking with the codebase convention.Use
error.name === "AbortError"to match the pattern used throughout the codebase, or refactor the idle wait to useAbortSignal.timeout()or similar standard async patterns:Proposed fix
} } catch (error) { - if (error instanceof Error && error.message === "AbortError") { + if (error instanceof Error && error.name === "AbortError") { this.logger.debug("Master queue consumer aborted", { loopId }); this.batchedSpanManager.cleanup(loopId); return;
🤖 Fix all issues with AI Agents
In @packages/redis-worker/src/fair-queue/index.ts:
- Around line 980-1002: When concurrencyManager.reserve(descriptor,
message.messageId) fails, you must release not only the current message but also
all remaining entries in claimedMessages back to the master queue instead of
breaking immediately; iterate the remaining claimedMessages (starting after the
current index) and call visibilityManager.release(message.messageId, queueId,
queueKey, queueItemsKey, masterQueueKey) for each to clear their inflight state
before stopping, then break so workerQueueManager.push continues only for
successfully reserved messages.
🧹 Nitpick comments (6)
packages/redis-worker/src/fair-queue/concurrency.ts (1)
117-135: Consider parallelizing Redis calls for better performance.The method logic is correct, but using
awaitin a loop serializes the Redis calls. For better performance, especially when there are multiple concurrency groups, consider usingPromise.allto fetch all group states in parallel.🔎 Proposed optimization using Promise.all
async getAvailableCapacity(queue: QueueDescriptor): Promise<number> { - let minCapacity = Infinity; - - for (const group of this.groups) { - const groupId = group.extractGroupId(queue); - const key = this.keys.concurrencyKey(group.name, groupId); - const current = await this.redis.scard(key); - const limit = (await group.getLimit(groupId)) || group.defaultLimit; - const available = Math.max(0, limit - current); - - minCapacity = Math.min(minCapacity, available); - } - - return minCapacity === Infinity ? 0 : minCapacity; + const capacities = await Promise.all( + this.groups.map(async (group) => { + const groupId = group.extractGroupId(queue); + const key = this.keys.concurrencyKey(group.name, groupId); + const current = await this.redis.scard(key); + const limit = (await group.getLimit(groupId)) || group.defaultLimit; + return Math.max(0, limit - current); + }) + ); + + return capacities.length > 0 ? Math.min(...capacities) : 0; }apps/webapp/app/env.server.ts (1)
954-959: Add validation for minimum values.These configuration values should be constrained to prevent invalid runtime behavior. Consider adding
.positive()or.min(1)to ensure:
BATCH_QUEUE_SHARD_COUNTis at least 1 (to avoid division by zero in shard mapping)BATCH_QUEUE_MASTER_QUEUE_LIMITis at least 1 (to ensure queues can be fetched)BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDSis at least 1 (to avoid potential blocking issues)🔎 Proposed validation constraints
- BATCH_QUEUE_SHARD_COUNT: z.coerce.number().int().default(1), + BATCH_QUEUE_SHARD_COUNT: z.coerce.number().int().positive().default(1), // Maximum queues to fetch from master queue per iteration - BATCH_QUEUE_MASTER_QUEUE_LIMIT: z.coerce.number().int().default(1000), + BATCH_QUEUE_MASTER_QUEUE_LIMIT: z.coerce.number().int().positive().default(1000), // Worker queue blocking timeout in seconds (for two-stage processing) - BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDS: z.coerce.number().int().default(10), + BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDS: z.coerce.number().int().positive().default(10),packages/redis-worker/src/fair-queue/index.ts (4)
976-977: Consider making worker queue ID configurableThe worker queue ID is hardcoded as
"worker-queue". If this system needs to support multiple independent worker queue pools in the future (e.g., for different priority levels or isolation), this would need refactoring.Consider extracting this as a configuration parameter:
// In constructor or options this.sharedWorkerQueueId = options.workerQueue?.queueId ?? "worker-queue";
1023-1027: Add abort signal check at loop startThe worker queue consumer loop doesn't check the abort signal at the start of the loop iteration (unlike the master queue loop at line 761-763). While the
blockingPopcall includes the abort signal, adding an early check would prevent unnecessary handler checks when already aborted.Suggested addition
try { while (this.isRunning) { + // Check abort signal + if (this.abortController.signal.aborted) { + break; + } + if (!this.messageHandler) { await new Promise((resolve) => setTimeout(resolve, this.consumerIntervalMs)); continue; }
759-918: Architectural observation: Two-stage vs direct processing pathsThe codebase now has two distinct processing paths:
- Two-stage (lines 759-918, 1015-1092): Master queue claims batches, pushes to shared worker queue, workers pop and process
- Direct (lines 1124-1271): Consumers claim and process messages directly, one at a time
The two-stage path uses batch claiming and numeric return values, while the direct path uses single-message claiming and boolean returns. This divergence is intentional based on the
workerQueueEnabledflag, but creates maintenance complexity.Consider documenting the trade-offs and decision criteria for when to use each mode, as this significantly impacts processing behavior, concurrency patterns, and fairness guarantees.
Also applies to: 1015-1092, 1124-1271
893-906: Move telemetry tracing outside the loop to reduce span overheadThe loop at lines 893-906 creates a separate trace span for each of the up to 10
recordProcessedcalls. Consider moving thetelemetry.tracewrapper outside the loop to wrap all records in a single span, which would reduce telemetry overhead during batch processing.For a more complete solution, the
FairSchedulerinterface could also be extended with arecordProcessedBatch(tenantId: string, queueId: string, count: number)method to allow schedulers to optimize their deficit tracking for batch operations.
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (11)
apps/webapp/app/env.server.tsapps/webapp/app/v3/runEngine.server.tsinternal-packages/run-engine/src/batch-queue/index.tsinternal-packages/run-engine/src/batch-queue/types.tsinternal-packages/run-engine/src/engine/index.tsinternal-packages/run-engine/src/engine/types.tspackages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/schedulers/drr.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/visibility.ts
🧰 Additional context used
📓 Path-based instructions (6)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
packages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/engine/types.tsinternal-packages/run-engine/src/engine/index.tsapps/webapp/app/v3/runEngine.server.tspackages/redis-worker/src/fair-queue/visibility.tsinternal-packages/run-engine/src/batch-queue/types.tsapps/webapp/app/env.server.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
packages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/engine/types.tsinternal-packages/run-engine/src/engine/index.tsapps/webapp/app/v3/runEngine.server.tspackages/redis-worker/src/fair-queue/visibility.tsinternal-packages/run-engine/src/batch-queue/types.tsapps/webapp/app/env.server.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
packages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/engine/types.tsinternal-packages/run-engine/src/engine/index.tsapps/webapp/app/v3/runEngine.server.tspackages/redis-worker/src/fair-queue/visibility.tsinternal-packages/run-engine/src/batch-queue/types.tsapps/webapp/app/env.server.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
{packages/core,apps/webapp}/**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use zod for validation in packages/core and apps/webapp
Files:
apps/webapp/app/v3/runEngine.server.tsapps/webapp/app/env.server.ts
apps/webapp/app/**/*.{ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/webapp.mdc)
Access all environment variables through the
envexport ofenv.server.tsinstead of directly accessingprocess.envin the Trigger.dev webapp
Files:
apps/webapp/app/v3/runEngine.server.tsapps/webapp/app/env.server.ts
apps/webapp/**/*.{ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/webapp.mdc)
apps/webapp/**/*.{ts,tsx}: When importing from@trigger.dev/corein the webapp, use subpath exports from the package.json instead of importing from the root path
Follow the Remix 2.1.0 and Express server conventions when updating the main trigger.dev webapp
Files:
apps/webapp/app/v3/runEngine.server.tsapps/webapp/app/env.server.ts
🧠 Learnings (3)
📓 Common learnings
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.trigger()` to trigger multiple different tasks at once from backend code
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.triggerByTask()` to batch trigger tasks by passing task instances for static task sets
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.triggerAndWait()` to batch trigger multiple different tasks and wait for results
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `batch.triggerByTaskAndWait()` to batch trigger tasks by passing task instances and wait for results
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `tasks.batchTrigger()` to trigger multiple runs of a single task with different payloads
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `yourTask.batchTrigger()` to trigger multiple runs of a task from inside another task
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `yourTask.batchTriggerAndWait()` to batch trigger tasks and wait for all results from a parent task
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
packages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/engine/types.tsinternal-packages/run-engine/src/engine/index.tsapps/webapp/app/v3/runEngine.server.tsinternal-packages/run-engine/src/batch-queue/types.tsinternal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:26:58.661Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-11-27T16:26:58.661Z
Learning: Applies to apps/webapp/app/**/*.{ts,tsx} : Access all environment variables through the `env` export of `env.server.ts` instead of directly accessing `process.env` in the Trigger.dev webapp
Applied to files:
apps/webapp/app/env.server.ts
🧬 Code graph analysis (5)
packages/redis-worker/src/fair-queue/concurrency.ts (1)
packages/redis-worker/src/fair-queue/types.ts (1)
QueueDescriptor(31-38)
internal-packages/run-engine/src/engine/index.ts (1)
internal-packages/run-engine/src/run-queue/index.ts (1)
options(1126-1142)
apps/webapp/app/v3/runEngine.server.ts (2)
apps/webapp/app/env.server.ts (1)
env(1291-1291)apps/supervisor/src/env.ts (1)
env(119-119)
packages/redis-worker/src/fair-queue/visibility.ts (2)
packages/redis-worker/src/fair-queue/keyProducer.ts (4)
queueKey(37-39)queueItemsKey(41-43)inflightKey(57-59)inflightDataKey(61-63)packages/redis-worker/src/fair-queue/types.ts (1)
InFlightMessage(207-218)
packages/redis-worker/src/fair-queue/index.ts (6)
packages/redis-worker/src/fair-queue/visibility.ts (2)
shardId(472-481)queueId(453-455)packages/redis-worker/src/fair-queue/concurrency.ts (1)
isAtCapacity(151-157)packages/redis-worker/src/fair-queue/telemetry.ts (1)
FairQueueAttributes(18-29)packages/redis-worker/src/fair-queue/masterQueue.ts (1)
queueId(195-197)packages/redis-worker/src/fair-queue/types.ts (1)
StoredMessage(62-79)packages/redis-worker/src/fair-queue/keyProducer.ts (3)
queueKey(37-39)queueItemsKey(41-43)masterQueueKey(29-31)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (17)
packages/redis-worker/src/fair-queue/types.ts (1)
514-515: LGTM!The optional
masterQueueLimitfield is well-documented and maintains backwards compatibility. The comment clearly indicates the default value.internal-packages/run-engine/src/engine/types.ts (1)
79-82: LGTM!The new optional configuration fields are well-documented and maintain backwards compatibility. The comments clearly describe their purpose.
internal-packages/run-engine/src/batch-queue/types.ts (2)
126-127: LGTM!The optional
masterQueueLimitfield is well-documented and maintains backwards compatibility.
201-204: LGTM!The new optional configuration fields are well-documented and maintain backwards compatibility. They align properly with the type definitions in other layers of the codebase.
apps/webapp/app/v3/runEngine.server.ts (1)
174-177: All three environment variables are properly defined inenv.server.tswith zod validation and appropriate defaults. They are correctly imported and accessed via theenvexport inrunEngine.server.ts. No action needed.packages/redis-worker/src/fair-queue/schedulers/drr.ts (1)
30-30: LGTM: Clean parameterization of master queue limit.The
masterQueueLimitfield is correctly introduced and used to replace the hardcoded value of 1000. The fallback default maintains backward compatibility.Also applies to: 39-39, 200-200
internal-packages/run-engine/src/engine/index.ts (1)
343-346: LGTM: Configuration propagation is clean and correct.The three new batch queue configuration options are properly threaded through to the BatchQueue initialization using optional chaining to handle undefined values gracefully.
internal-packages/run-engine/src/batch-queue/index.ts (3)
111-111: LGTM: DRR masterQueueLimit correctly forwarded.The
masterQueueLimitis properly passed through to the DRRScheduler configuration.
125-125: LGTM: Safe shard count default.The
shardCountconfiguration correctly uses the provided value with a safe fallback to 1, maintaining backward compatibility.
137-142: LGTM: Opt-in worker queue configuration.The conditional worker queue configuration is a clean approach for enabling two-stage processing only when explicitly configured. The
blockingTimeoutSecondsparameter correctly gates the feature.packages/redis-worker/src/fair-queue/visibility.ts (3)
142-215: LGTM: Well-structured batch claim implementation.The
claimBatchmethod is correctly implemented:
- Mirrors the single-message
claimmethod structure for consistency- Properly handles the paired result array from the Lua script
- Includes appropriate error handling for corrupted payloads
- Logs batch operations for observability
The error handling correctly removes corrupted messages from in-flight tracking to prevent leaks.
529-577: LGTM: Atomic batch claim Lua script is correct.The
claimMessageBatchLua script correctly implements atomic batch message claiming:
- Uses
ZRANGEwith correct 0-based indexing to fetch up tomaxCountmessages- Atomically removes messages from the queue and adds them to in-flight tracking
- Gracefully handles missing payloads by removing orphaned queue entries
- Returns results in the expected flat array format
The atomicity guarantees prevent race conditions between concurrent consumers.
658-667: LGTM: TypeScript type definition is correct.The TypeScript type definition for
claimMessageBatchaccurately reflects the Lua script's signature and return type.packages/redis-worker/src/fair-queue/index.ts (4)
298-308: LGTM: Explicit undefined assignmentThe explicit
undefinedassignment forworkerQueuewhen the resolver is not provided makes the intent clearer without changing behavior.Also applies to: 394-404
866-874: LGTM: Tenant-level capacity checkThe tenant capacity check before processing queues prevents unnecessary work when a tenant is at capacity, and correctly skips all remaining queues for that tenant. This is an effective optimization.
Based on learnings, batch processing should respect tenant-level concurrency limits.
961-963: Verified: claimBatch method properly implemented with comprehensive edge case handlingThe
claimBatchmethod exists in VisibilityManager and handles all relevant edge cases. The implementation uses a Lua script for atomic operations and properly handles scenarios where fewer messages are claimed than requested, missing data, JSON parsing errors, and empty results. The method includes error logging and cleanup for corrupted messages.
937-946: Verified: All method implementations are correctThe code properly calls
getAvailableCapacity()to determine batch size limits based on available concurrency capacity. All referenced methods—getAvailableCapacity(),claimBatch(),reserve(), andrelease()—are correctly implemented in their respective manager classes. The logic for handling partial batch processing when concurrency limits are reached is sound.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
ericallam
force-pushed
the
ea-branch-113-3
branch
from
08f635d to
797ed36
Compare
PR Review: fix(batch): optimize processing batch trigger v2This PR introduces significant performance optimizations to the batch queue processing system. Here's my detailed review: ✅ Strengths1. Batch Claiming Architecture 2. Deficit Round Robin Batch Support 3. Shared Worker Queue Pattern 4. Adaptive Loop Timing 5. Comprehensive Test Coverage
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI Agents
In @internal-packages/run-engine/src/batch-queue/index.ts:
- Around line 135-142: The current truthiness check on
options.workerQueueBlockingTimeoutSeconds disables workerQueue when the value is
0; change the condition to explicitly check for undefined (e.g.,
options.workerQueueBlockingTimeoutSeconds !== undefined) so 0 is treated as a
valid blockingTimeoutSeconds; update the object creation for workerQueue
(referencing workerQueue, options.workerQueueBlockingTimeoutSeconds, and
blockingTimeoutSeconds) to use this explicit undefined check.
In @packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts:
- Around line 880-883: The test title is misleading: the test named "should
release remaining claimed messages when concurrency reservation fails" does not
simulate a reservation failure or check release behavior but instead verifies
successful completion under concurrency limits; rename the test title in the
describe/redisTest invocation to accurately reflect the behavior being asserted
(e.g., "should process all messages respecting concurrency limits in two-stage
mode" or "should complete batch processing without leaking in-flight messages")
so the name matches the assertions in this test block and update any related
comments to match.
🧹 Nitpick comments (4)
packages/redis-worker/src/fair-queue/index.ts (2)
809-814: Inconsistent abort error detection pattern.Line 810 checks
error.message === "AbortError", but line 1109 in#runWorkerQueueConsumerLoopcheckserror.name === "AbortError". The timeout rejection on line 797 creates an error with message"AbortError", so this works, but the inconsistency could cause confusion.🔎 Consider using consistent abort error detection
} catch (error) { - if (error instanceof Error && error.message === "AbortError") { + if (error instanceof Error && (error.name === "AbortError" || error.message === "AbortError")) { this.logger.debug("Master queue consumer aborted", { loopId });Or create a helper function for consistent abort detection across all loops.
912-924: Telemetry fallback loop is inefficient for large batches.When
recordProcessedBatchis not available, the fallback iterates and traces each message individually. For large batch sizes, this creates significant tracing overhead.🔎 Consider batching the fallback telemetry
} else if (this.scheduler.recordProcessed) { + // Fallback: record each message individually + // Note: This is less efficient than batch variant for (let i = 0; i < processedFromQueue; i++) { - await this.telemetry.trace( - "recordProcessed", - async (span) => { - span.setAttribute(FairQueueAttributes.QUEUE_ID, queueId); - span.setAttribute(FairQueueAttributes.TENANT_ID, tenantId); - await this.scheduler.recordProcessed!(tenantId, queueId); - }, - { kind: SpanKind.INTERNAL } - ); + await this.scheduler.recordProcessed!(tenantId, queueId); } }Creating a span per message in the fallback path may not add value since the messages are identical operations.
packages/redis-worker/src/fair-queue/visibility.ts (1)
686-686: Use#operator instead of deprecatedtable.getnin Lua.
table.getnwas deprecated in Lua 5.1 and removed in later versions. Redis uses Lua 5.1, so this works, but#ARGVis the idiomatic approach and is more consistent with the rest of the codebase (e.g., line 594 uses#items).🔎 Suggested fix
-- Remaining args are: members..., messageIds... -- Calculate how many messages we have -local numMessages = (table.getn(ARGV) - 2) / 2 +local numMessages = (#ARGV - 2) / 2 local membersStart = 3internal-packages/run-engine/src/batch-queue/index.ts (1)
125-125: Good: Dynamic shardCount with sensible default.The change from a hardcoded value to
options.shardCount ?? 1improves configurability while maintaining backwards compatibility.Consider adding validation for the shardCount value
While not critical, you might want to validate that
shardCountis a positive integer to catch configuration errors early:- shardCount: options.shardCount ?? 1, + shardCount: Math.max(1, Math.floor(options.shardCount ?? 1)),Or add validation in the constructor with a more descriptive error:
if (options.shardCount !== undefined && (options.shardCount < 1 || !Number.isInteger(options.shardCount))) { throw new Error(`Invalid shardCount: ${options.shardCount}. Must be a positive integer.`); }
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (16)
apps/webapp/app/env.server.tsapps/webapp/app/v3/runEngine.server.tsinternal-packages/run-engine/src/batch-queue/index.tsinternal-packages/run-engine/src/batch-queue/types.tsinternal-packages/run-engine/src/engine/index.tsinternal-packages/run-engine/src/engine/types.tspackages/redis-worker/src/fair-queue/concurrency.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/scheduler.tspackages/redis-worker/src/fair-queue/schedulers/drr.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.tspackages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/types.tspackages/redis-worker/src/fair-queue/visibility.ts
🚧 Files skipped from review as they are similar to previous changes (7)
- internal-packages/run-engine/src/engine/index.ts
- apps/webapp/app/v3/runEngine.server.ts
- apps/webapp/app/env.server.ts
- internal-packages/run-engine/src/batch-queue/types.ts
- internal-packages/run-engine/src/engine/types.ts
- packages/redis-worker/src/fair-queue/concurrency.ts
- packages/redis-worker/src/fair-queue/types.ts
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
packages/redis-worker/src/fair-queue/scheduler.tspackages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.tspackages/redis-worker/src/fair-queue/visibility.tspackages/redis-worker/src/fair-queue/index.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
packages/redis-worker/src/fair-queue/scheduler.tspackages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.tspackages/redis-worker/src/fair-queue/visibility.tspackages/redis-worker/src/fair-queue/index.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
packages/redis-worker/src/fair-queue/scheduler.tspackages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.tspackages/redis-worker/src/fair-queue/visibility.tspackages/redis-worker/src/fair-queue/index.ts
**/*.{test,spec}.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use vitest for all tests in the Trigger.dev repository
Files:
packages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.ts
**/*.test.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.test.{ts,tsx,js,jsx}: Test files should live beside the files under test and use descriptivedescribeanditblocks
Avoid mocks or stubs in tests; use helpers from@internal/testcontainerswhen Redis or Postgres are needed
Use vitest for unit tests
Files:
packages/redis-worker/src/fair-queue/tests/drr.test.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/visibility.test.tspackages/redis-worker/src/fair-queue/tests/concurrency.test.ts
🧠 Learnings (1)
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/schedulers/drr.tsinternal-packages/run-engine/src/batch-queue/index.ts
🧬 Code graph analysis (3)
packages/redis-worker/src/fair-queue/tests/drr.test.ts (3)
packages/redis-worker/src/fair-queue/schedulers/drr.ts (3)
keys(198-201)redis(275-345)DRRScheduler(25-346)packages/redis-worker/src/fair-queue/keyProducer.ts (1)
DefaultFairQueueKeyProducer(16-133)packages/redis-worker/src/fair-queue/schedulers/index.ts (1)
DRRScheduler(5-5)
packages/redis-worker/src/fair-queue/tests/visibility.test.ts (2)
packages/redis-worker/src/fair-queue/visibility.ts (3)
VisibilityManager(29-745)redis(534-744)queueId(504-506)packages/redis-worker/src/fair-queue/index.ts (4)
redis(1826-1916)queueId(1737-1750)queueId(1752-1786)queueId(1788-1790)
packages/redis-worker/src/fair-queue/tests/concurrency.test.ts (1)
packages/redis-worker/src/fair-queue/types.ts (1)
QueueDescriptor(31-38)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (22)
packages/redis-worker/src/fair-queue/scheduler.ts (1)
31-38: LGTM! Well-documented batch processing hook.The
recordProcessedBatchmethod is properly designed as a batch-optimized variant ofrecordProcessed. The documentation clearly explains its purpose (reducing Redis calls when processing multiple messages), and the default no-op implementation follows the established pattern for base scheduler hooks.packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts (1)
884-980: Test coverage for two-stage processing looks comprehensive.The test properly validates:
- Two-stage worker queue with concurrency limits
- All messages are processed successfully
- No messages remain in-flight after completion
- Proper cleanup occurs
packages/redis-worker/src/fair-queue/tests/concurrency.test.ts (1)
225-384: Excellent test coverage for getAvailableCapacity.The test suite comprehensively validates capacity calculation across multiple scenarios:
- Single group capacity and dynamic updates after reservations
- Minimum capacity calculation across multiple groups (correctly returns the bottleneck)
- Edge case: zero capacity when any group is at limit
- Edge case: zero capacity when no groups are configured
All tests properly clean up resources and follow vitest best practices.
packages/redis-worker/src/fair-queue/tests/visibility.test.ts (2)
248-448: Excellent batch claim test coverage with robust edge case handling.The
claimBatchtest suite is comprehensive and includes:
- Atomic batch operations with proper count verification
- Empty queue handling
- Partial batch behavior when fewer messages available
- Particularly good: The corrupted message test (lines 381-447) validates that invalid JSON is gracefully skipped while valid messages continue processing
This defensive approach prevents one corrupted message from blocking an entire batch.
450-599: Comprehensive batch release test coverage.The
releaseBatchtests properly validate:
- Atomic release of multiple messages with correct state updates
- Graceful handling of empty input (no exceptions thrown)
- Master queue timestamp updates with oldest message
All tests include proper verification of both in-flight and queue counts, ensuring state consistency.
packages/redis-worker/src/fair-queue/tests/drr.test.ts (1)
118-174: LGTM! Proper test coverage for batch deficit decrement.The new tests validate
recordProcessedBatchbehavior:
- Normal batch decrement (15 - 7 = 8)
- Floor-at-zero when decrementing more than available (3 - 10 = 0)
Both tests follow the established pattern for deficit management tests and properly verify boundary conditions.
packages/redis-worker/src/fair-queue/schedulers/drr.ts (4)
30-39: LGTM! Configurable master queue limit.The addition of
masterQueueLimitas a configurable field with a sensible default of 1000 improves flexibility over the previously hardcoded value.
143-153: LGTM! Batch deficit decrement implementation.The
recordProcessedBatchmethod correctly follows the existing pattern ofrecordProcessedwhile supporting batch operations. The delegation to#decrementDeficitBatchis clean.
264-273: LGTM! Atomic batch deficit decrement with proper floor.The Lua script correctly:
- Decrements the deficit by the provided count atomically
- Floors at 0 to prevent negative deficits
- Returns the new deficit value as a string
This mirrors the single-message
drrDecrementDeficitlogic appropriately.Also applies to: 325-344
359-364: LGTM! Interface extension for batch command.The
drrDecrementDeficitBatchinterface declaration correctly matches the Lua command implementation.packages/redis-worker/src/fair-queue/index.ts (6)
87-87: LGTM! Batch claim size configuration.The
batchClaimSizefield with a default of 10 provides reasonable control over batch processing throughput.Also applies to: 143-144
794-806: Memory leak fix correctly implemented.The abort listener is now properly removed when the timeout fires (line 802), preventing listener accumulation during idle periods. This addresses the previously flagged memory leak issue.
872-880: LGTM! Per-tenant capacity check optimization.Checking tenant capacity before processing queues and breaking early prevents unnecessary work when the tenant is at capacity. The stat tracking with
tenant_capacity_skippedaids observability.
1010-1020: Stranded messages fix correctly implemented.When concurrency reservation fails, all remaining claimed messages (from index
ionward) are now released back to the queue viareleaseBatch. This prevents messages from being stranded in the in-flight set until visibility timeout expires. Good fix for the previously flagged issue.
993-999: Shared worker queue pattern is acceptable with documented trade-offs.The comments clearly document the trade-off: simpler code and fair distribution vs. potential contention under very high load. The suggestion to add optional sharding in the future is noted.
311-312: LGTM! Consistent undefined handling for workerQueue.Setting
workerQueuetoundefinedwhen no resolver is provided (instead of potentially calling the resolver with incomplete data) is correct.Also applies to: 407-408
packages/redis-worker/src/fair-queue/visibility.ts (5)
130-214: LGTM! Batch claim implementation with proper error handling.The
claimBatchmethod correctly:
- Handles empty results
- Parses alternating messageId/payload pairs
- Handles JSON parse errors gracefully by removing corrupted messages from in-flight
- Provides appropriate logging
313-363: LGTM! Batch release implementation.The
releaseBatchmethod correctly:
- Returns early on empty input
- Builds parallel arrays of members and messageIds for the Lua script
- Provides appropriate logging
579-627: LGTM! Atomic batch claim Lua script.The script correctly handles the batch claim operation:
- Fetches up to
maxCountmessages usingZRANGE- Processes each message atomically (remove from queue, add to in-flight)
- Handles missing payload data gracefully
- Returns alternating messageId/payload pairs
671-721: LGTM! Atomic batch release Lua script with single master queue update.The script correctly:
- Calculates array positions for members and messageIds
- Releases each message atomically
- Updates the master queue only once at the end (performance optimization)
- Returns the count of released messages
760-768: LGTM! Redis interface extensions for batch operations.The interface declarations correctly match the Lua command implementations for both
claimMessageBatchandreleaseMessageBatch.Also applies to: 782-791
internal-packages/run-engine/src/batch-queue/index.ts (1)
111-111: The masterQueueLimit forwarding is correct.The property is properly defined as optional in
DRRConfig(line 192 of types.ts), andDRRSchedulercorrectly defaults undefined values to 1000 (line 39 of drr.ts). The forwarding at line 111 is safe and type-correct.
| // Enable two-stage processing with worker queues for better parallelism (when configured) | ||
| // Worker queues provide better concurrency by separating queue selection from message processing | ||
| workerQueue: options.workerQueueBlockingTimeoutSeconds | ||
| ? { | ||
| enabled: true, | ||
| blockingTimeoutSeconds: options.workerQueueBlockingTimeoutSeconds, | ||
| } | ||
| : undefined, |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check BatchQueueOptions type definition
echo "=== BatchQueueOptions type definition ==="
if [ -f "internal-packages/run-engine/src/batch-queue/types.ts" ]; then
cat -n internal-packages/run-engine/src/batch-queue/types.ts | head -100
else
echo "File not found, searching for it..."
fd "types.ts" internal-packages/run-engine/src/batch-queue/
fiRepository: triggerdotdev/trigger.dev
Length of output: 4666
🏁 Script executed:
#!/bin/bash
# Search for workerQueueBlockingTimeoutSeconds usage and documentation
echo "=== Searching for workerQueueBlockingTimeoutSeconds ==="
rg -n "workerQueueBlockingTimeoutSeconds" --type ts -B 3 -A 3
echo -e "\n=== Searching for blockingTimeoutSeconds in type definitions ==="
rg -n "blockingTimeoutSeconds" --type ts -B 3 -A 3Repository: triggerdotdev/trigger.dev
Length of output: 8707
🏁 Script executed:
#!/bin/bash
# Check the actual file to see context around lines 135-142
echo "=== Context around lines 135-142 in index.ts ==="
sed -n '125,150p' internal-packages/run-engine/src/batch-queue/index.ts | cat -nRepository: triggerdotdev/trigger.dev
Length of output: 1497
Use explicit undefined check to allow 0 as a valid timeout value.
The truthiness check on options.workerQueueBlockingTimeoutSeconds prevents 0 from being used as a valid timeout value. In timeout contexts (Redis BLPOP), 0 is valid and typically means block indefinitely. If the timeout is set via environment variable to "0", it will be coerced to the number 0 and incorrectly disable the worker queue.
Change the condition to:
- workerQueue: options.workerQueueBlockingTimeoutSeconds
+ workerQueue: options.workerQueueBlockingTimeoutSeconds !== undefined
? {
enabled: true,
blockingTimeoutSeconds: options.workerQueueBlockingTimeoutSeconds,
}
: undefined,🤖 Prompt for AI Agents
In @internal-packages/run-engine/src/batch-queue/index.ts around lines 135 -
142, The current truthiness check on options.workerQueueBlockingTimeoutSeconds
disables workerQueue when the value is 0; change the condition to explicitly
check for undefined (e.g., options.workerQueueBlockingTimeoutSeconds !==
undefined) so 0 is treated as a valid blockingTimeoutSeconds; update the object
creation for workerQueue (referencing workerQueue,
options.workerQueueBlockingTimeoutSeconds, and blockingTimeoutSeconds) to use
this explicit undefined check.
Code Review: PR #2841 - fix(batch): optimize processing batch trigger v2Thank you for this PR! This is a substantial optimization to the batch queue processing system, introducing batch claiming, worker queue consolidation, and improved concurrency handling. Here's my detailed review: ✅ Strengths
|
Code Review: fix(batch): optimize processing batch trigger v2SummaryThis PR refactors the FairQueue and BatchQueue to implement a two-stage message processing architecture. FairQueue now handles fair scheduling and routes messages to worker queues, while external consumers (like BatchQueue) run their own worker queue consumer loops. This is a significant architectural change that improves separation of concerns and enables more flexible scaling. 👍 Strengths
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In @packages/redis-worker/src/fair-queue/index.ts:
- Line 986: The call to workerQueueResolver is passing message.payload instead
of the full StoredMessage, causing a type/runtime mismatch; update the
invocation to pass the complete message object (the StoredMessage instance
containing messageId, queueId, tenantId, payload, etc.) rather than
message.payload so workerQueueResolver receives the expected shape (same pattern
used in the enqueue methods and as defined by the StoredMessage type).
In @packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts:
- Around line 1081-1084: The test name is misleading; inside the describe
"two-stage processing with concurrency limits" update the redisTest invocation's
first argument (the test title string) to reflect the actual behavior being
validated (successful processing under concurrency limits) — e.g., change the
title passed to redisTest from "should release remaining claimed messages when
concurrency reservation fails" to "should process all messages respecting
concurrency limits in two-stage mode" (or "should complete batch processing
without leaking in-flight messages") so the test name matches the scenario.
🧹 Nitpick comments (1)
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts (1)
22-210: Consider extracting TestFairQueueHelper to a shared test utility.The TestFairQueueHelper class is duplicated across both
raceConditions.test.tsandfairQueue.test.ts. Extracting it to a shared test utility file would reduce duplication and improve maintainability.♻️ Suggested refactor
Create a new file
packages/redis-worker/src/fair-queue/tests/helpers.ts:import { createRedisClient, type RedisOptions } from "@internal/redis"; import { FairQueue, WorkerQueueManager, type FairQueueOptions, type StoredMessage, type FairQueueKeyProducer, } from "../index.js"; import { z } from "zod"; export const TEST_WORKER_QUEUE_ID = "test-worker-queue"; export class TestFairQueueHelper<TSchema extends z.ZodTypeAny> { // ... (move the entire class implementation here) }Then import in both test files:
import { TestFairQueueHelper, TEST_WORKER_QUEUE_ID } from "./helpers.js";
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
internal-packages/run-engine/src/batch-queue/index.tsinternal-packages/run-engine/src/batch-queue/types.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.tspackages/redis-worker/src/fair-queue/types.ts
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
internal-packages/run-engine/src/batch-queue/types.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/types.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
internal-packages/run-engine/src/batch-queue/types.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/types.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
internal-packages/run-engine/src/batch-queue/types.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/fair-queue/types.ts
**/*.{test,spec}.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use vitest for all tests in the Trigger.dev repository
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.ts
**/*.test.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.test.{ts,tsx,js,jsx}: Test files should live beside the files under test and use descriptivedescribeanditblocks
Avoid mocks or stubs in tests; use helpers from@internal/testcontainerswhen Redis or Postgres are needed
Use vitest for unit tests
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.ts
🧠 Learnings (8)
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
internal-packages/run-engine/src/batch-queue/types.tspackages/redis-worker/src/fair-queue/tests/fairQueue.test.tspackages/redis-worker/src/fair-queue/tests/raceConditions.test.tsinternal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/types.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger.config.ts : Configure OpenTelemetry instrumentations and exporters in trigger.config.ts for enhanced logging
Applied to files:
internal-packages/run-engine/src/batch-queue/types.ts
📚 Learning: 2025-11-14T16:03:06.917Z
Learnt from: matt-aitken
Repo: triggerdotdev/trigger.dev PR: 2681
File: apps/webapp/app/services/platform.v3.server.ts:258-302
Timestamp: 2025-11-14T16:03:06.917Z
Learning: In `apps/webapp/app/services/platform.v3.server.ts`, the `getDefaultEnvironmentConcurrencyLimit` function intentionally throws an error (rather than falling back to org.maximumConcurrencyLimit) when the billing client returns undefined plan limits. This fail-fast behavior prevents users from receiving more concurrency than their plan entitles them to. The org.maximumConcurrencyLimit fallback is only for self-hosted deployments where no billing client exists.
Applied to files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
📚 Learning: 2025-11-27T16:27:48.109Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: AGENTS.md:0-0
Timestamp: 2025-11-27T16:27:48.109Z
Learning: Applies to **/*.test.{ts,tsx,js,jsx} : Avoid mocks or stubs in tests; use helpers from `internal/testcontainers` when Redis or Postgres are needed
Applied to files:
packages/redis-worker/src/fair-queue/tests/raceConditions.test.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `runs.subscribeToBatch()` to subscribe to changes for all runs in a batch
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-08-14T18:35:44.370Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: apps/webapp/app/env.server.ts:764-765
Timestamp: 2025-08-14T18:35:44.370Z
Learning: The BoolEnv helper in apps/webapp/app/utils/boolEnv.ts uses z.preprocess with inconsistent default value types across the codebase - some usages pass boolean defaults (correct) while others pass string defaults (incorrect), leading to type confusion. The helper should enforce boolean-only defaults or have clearer documentation.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-08-14T10:09:02.528Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: internal-packages/run-engine/src/engine/index.ts:466-467
Timestamp: 2025-08-14T10:09:02.528Z
Learning: In the triggerdotdev/trigger.dev codebase, it's acceptable to pass `string | undefined` types directly to Prisma operations (both create and update). The codebase consistently uses this pattern and the team is comfortable with how Prisma handles undefined values.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:26:58.661Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-11-27T16:26:58.661Z
Learning: Use `trigger.dev/redis-worker` for background job and worker system needs in the webapp and run engine
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
🧬 Code graph analysis (1)
packages/redis-worker/src/fair-queue/index.ts (7)
packages/redis-worker/src/fair-queue/workerQueue.ts (1)
WorkerQueueManager(26-275)packages/redis-worker/src/fair-queue/types.ts (1)
StoredMessage(62-79)packages/redis-worker/src/fair-queue/visibility.ts (3)
shardId(523-532)queueId(504-506)messageId(508-510)packages/redis-worker/src/fair-queue/keyProducer.ts (4)
masterQueueKey(29-31)queueKey(37-39)queueItemsKey(41-43)inflightDataKey(61-63)packages/redis-worker/src/fair-queue/concurrency.ts (1)
isAtCapacity(170-176)packages/redis-worker/src/fair-queue/telemetry.ts (1)
FairQueueAttributes(18-29)packages/redis-worker/src/fair-queue/masterQueue.ts (1)
queueId(195-197)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (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 (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (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 (ubuntu-latest - pnpm)
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: Cursor Bugbot
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (15)
internal-packages/run-engine/src/batch-queue/types.ts (1)
126-128: LGTM! Configuration options are well-documented.The new configuration fields are properly typed with clear documentation:
masterQueueLimitfor controlling batch sizes from master queueshardCountfor horizontal scalingworkerQueueBlockingTimeoutSecondsenabling two-stage processing- Consumer trace rotation controls for better observability
All fields are optional with sensible defaults documented in comments.
Also applies to: 201-204, 225-228
internal-packages/run-engine/src/batch-queue/index.ts (3)
455-510: Excellent lifecycle management with proper idempotency and cleanup.The refactored lifecycle methods properly handle:
- Idempotent start/stop with
isRunningguard- Graceful shutdown via AbortController
- Sequential cleanup of all resources (FairQueue → WorkerQueueManager → CompletionTracker → Redis)
- Safe waiting for worker loops with
Promise.allSettled
595-671: Worker queue consumer loop is well-structured.The implementation follows best practices:
- Uses BatchedSpanManager for efficient tracing with rotation
- Properly handles abort signals for graceful shutdown
- Tracks metrics (empty iterations, invalid keys, messages processed)
- Cleanup in finally block ensures span manager resources are released
677-879: Message handling properly integrates with worker queue architecture.The refactored message handler correctly:
- Retrieves in-flight message data from FairQueue storage
- Calculates accurate queue time metrics from stored timestamp
- Ensures message completion in all paths (success, failure, unexpected error)
- Preserves idempotency through itemIndex tracking
- Maintains proper completion callback error handling
packages/redis-worker/src/fair-queue/types.ts (4)
189-194: Good addition of batch processing support to scheduler interface.The
recordProcessedBatchmethod enables efficient batch updates to scheduler state, reducing Redis calls when processing multiple messages. Being optional, it allows backwards compatibility with a fallback to individualrecordProcessedcalls.
380-382: Well-designed batch processing configuration options.The new optional fields support batch processing optimizations:
batchClaimSizecontrols batch claiming for efficiencymasterQueueLimitcaps queue fetching across schedulersAll are optional with clear documentation, maintaining backwards compatibility.
Also applies to: 529-530, 582-582
353-358: Breaking change: Worker queue configuration now required.The
workerQueuefield inFairQueueOptionsis now required (was optional). This enforces the two-stage processing model where FairQueue routes messages to worker queues. All FairQueue instantiations in the codebase have been updated to provide this configuration.
292-297: Usetypeinstead ofinterfaceand address type signature mismatch.
WorkerQueueOptionsshould usetypeinstead ofinterfaceper TypeScript guidelines. Additionally, theresolveWorkerQueuetype signature expects(message: StoredMessage<TPayload>) => string, but implementations use() => string. Line 986 passes onlymessage.payloadwhile lines 286 and 382 construct and pass the fullStoredMessageobject. Align the type definition with actual usage patterns across the codebase.Likely an incorrect or invalid review comment.
packages/redis-worker/src/fair-queue/index.ts (7)
757-769: Memory leak fix appears correct.The manual listener removal on line 765 addresses the past review concern about event listener accumulation. When the timeout fires, the abort listener is explicitly removed, preventing unbounded accumulation. The
{ once: true }option handles removal when abort fires first.
970-982: Batch release logic correctly addresses past review concern.The implementation now properly releases all remaining unclaimed messages when concurrency reservation fails (lines 973-979). Using
releaseBatchwithclaimedMessages.slice(i)ensures messages aren't stranded in the inflight state, addressing the critical issue flagged in previous reviews.
1002-1203: Public API methods are well-structured.The new lifecycle methods (getMessageData, heartbeatMessage, completeMessage, releaseMessage, failMessage) provide a clean external consumer interface. The consistent pattern of retrieving message data, performing operations, releasing concurrency, and cleaning up caches is correct and maintainable.
840-848: Tenant capacity check is a good optimization.Checking tenant capacity before attempting to claim messages prevents unnecessary work when the tenant is already at its concurrency limit. The early break correctly skips remaining queues for that tenant, reducing contention and improving efficiency.
867-892: Batch recording implementation is efficient.The logic prefers the batch variant (
recordProcessedBatch) when available, falling back to individual calls for backward compatibility. This approach minimizes scheduler overhead while maintaining compatibility with schedulers that don't implement batch operations.
923-932: Available capacity calculation is correct.The logic properly determines the claim size based on available concurrency capacity, capping it at
batchClaimSize. The cooloff increment when capacity is zero prevents repeated futile claim attempts, which is good for performance.
726-770: hadWork-based adaptive waiting is efficient.The pattern of tracking whether work was performed and adjusting wait times accordingly (1ms when busy,
consumerIntervalMswhen idle) optimizes both throughput and CPU usage. This prevents busy-waiting while maintaining responsiveness when messages are available.Also applies to: 787-904
| // Determine worker queue | ||
| const workerQueueId = message.payload.workerQueue ?? queueId; | ||
| // Resolve which worker queue this message should go to | ||
| const workerQueueId = this.workerQueueResolver(message.payload); |
There was a problem hiding this comment.
Critical: Wrong parameter passed to workerQueueResolver.
The workerQueueResolver expects a StoredMessage (line 85), but message.payload is being passed instead of the full message object. This type mismatch will cause runtime errors since the resolver likely needs access to fields like messageId, queueId, and tenantId.
Compare with correct usage in enqueue methods (lines 285-295, 381-391) where the full message object is constructed and passed.
🐛 Proposed fix
- // Resolve which worker queue this message should go to
- const workerQueueId = this.workerQueueResolver(message.payload);
+ // Resolve which worker queue this message should go to
+ const workerQueueId = this.workerQueueResolver(message);🤖 Prompt for AI Agents
In @packages/redis-worker/src/fair-queue/index.ts at line 986, The call to
workerQueueResolver is passing message.payload instead of the full
StoredMessage, causing a type/runtime mismatch; update the invocation to pass
the complete message object (the StoredMessage instance containing messageId,
queueId, tenantId, payload, etc.) rather than message.payload so
workerQueueResolver receives the expected shape (same pattern used in the
enqueue methods and as defined by the StoredMessage type).
| describe("two-stage processing with concurrency limits", () => { | ||
| redisTest( | ||
| "should release remaining claimed messages when concurrency reservation fails", | ||
| { timeout: 30000 }, |
There was a problem hiding this comment.
Test name doesn't match the scenario being tested.
The test name "should release remaining claimed messages when concurrency reservation fails" suggests testing a failure/cleanup scenario, but the test actually validates successful processing of all messages under concurrency limits without simulating any reservation failure.
Consider renaming to:
- "should process all messages respecting concurrency limits in two-stage mode"
- "should complete batch processing without leaking in-flight messages"
🤖 Prompt for AI Agents
In @packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts around lines
1081 - 1084, The test name is misleading; inside the describe "two-stage
processing with concurrency limits" update the redisTest invocation's first
argument (the test title string) to reflect the actual behavior being validated
(successful processing under concurrency limits) — e.g., change the title passed
to redisTest from "should release remaining claimed messages when concurrency
reservation fails" to "should process all messages respecting concurrency limits
in two-stage mode" (or "should complete batch processing without leaking
in-flight messages") so the test name matches the scenario.
PR Review: fix(batch): optimize processing batch trigger v2SummaryThis PR refactors the batch processing architecture to use a two-stage processing model where FairQueue handles fair scheduling and routes messages to worker queues, while BatchQueue runs its own consumer loops to process messages from those worker queues. This is a significant architectural change aimed at optimizing batch trigger processing. 🟢 Strengths1. Clean Architectural SeparationThe separation of concerns between FairQueue (scheduling/routing) and external consumers (processing) is well-designed. FairQueue now focuses on fair scheduling via DRR and message routing, while consumers handle actual processing. This makes the system more modular and easier to reason about. 2. Comprehensive Test CoverageThe PR includes extensive test updates and additions:
3. Performance Optimizations
4. Good Use of Atomic OperationsThe Lua scripts for atomic operations are well-implemented:
5. Configurable ShardingNew environment variables ( 🟡 Suggestions for Improvement1. Single Worker Queue Potential BottleneckIn const workerQueueId = "worker-queue";While the comment acknowledges this trade-off, consider documenting performance thresholds more precisely or adding metrics to monitor this potential bottleneck. Suggestion: Add a gauge metric to track worker queue depth for proactive monitoring. 2. Error Handling in Consumer LoopIn } catch (error) {
if (this.abortController.signal.aborted) {
break;
}
this.logger.error("Worker queue consumer error", {
loopId,
error: error instanceof Error ? error.message : String(error),
});
this.batchedSpanManager.markForRotation(loopId);
}Consider adding:
3. Message Key ParsingThe message key format const colonIndex = messageKey.indexOf(":");
if (colonIndex === -1) {
this.logger.error("Invalid message key format", { messageKey });
return false;
}Suggestion: Consider using a more robust delimiter or encoding scheme, or document the constraint that messageIds must not contain colons. 4. Memory Management for Cooloff StatesGood that if (this.queueCooloffStates.size >= this.maxCooloffStatesSize) {
this.logger.warn("Cooloff states cache hit size cap, clearing all entries"...);
this.queueCooloffStates.clear();
}Suggestion: Consider LRU eviction instead of full clear. 5. TypeScript Strict Null ChecksThe non-null assertions in const available = Math.max(0, limits[i]! - currents[i]!);While safe in context, could be more defensive. 🔴 Potential Issues1. Removed
|
| Area | Coverage |
|---|---|
| Basic enqueue/process | ✅ Good |
| Fair scheduling (DRR) | ✅ Good |
| Visibility timeout/reclaim | ✅ Good |
| Concurrency limiting | ✅ Good |
| Batch operations | ✅ Good |
| Worker queue integration | ✅ Good |
| Race conditions | ✅ Good |
| Error handling |
📝 Minor Suggestions
-
Documentation: The comment block explaining the two-stage architecture is helpful. Consider adding an architecture diagram to the README.
-
Naming:
BATCH_WORKER_QUEUE_ID = "batch-worker-queue"vsworkerQueueId = "worker-queue"- consider consistent naming across the codebase. -
Magic Numbers: The
1msdelay when there's work (const waitMs = hadWork ? 1 : this.consumerIntervalMs) could be a named constant for clarity.
✅ Overall Assessment
This is a well-structured refactoring that improves the batch processing architecture. The code is clean, well-tested, and follows good practices for distributed systems (atomic operations, visibility timeouts, fair scheduling). The separation of scheduling from processing is a good architectural decision that will make the system more maintainable.
Recommendation: Approve with minor suggestions addressed.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In @packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts:
- Line 1094: The test title "should release remaining claimed messages when
concurrency reservation fails" is misleading relative to assertions that verify
successful processing of 5 messages with concurrency 2; update the test name
string in the test declaration (the describe/it/test call in fairQueue.test.ts)
to reflect the actual behavior such as "should process all messages respecting
concurrency limits in two-stage mode" or "should complete batch processing
without leaking messages under concurrency limits" so the test name matches the
validated scenario.
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
**/*.{test,spec}.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use vitest for all tests in the Trigger.dev repository
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
**/*.test.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.test.{ts,tsx,js,jsx}: Test files should live beside the files under test and use descriptivedescribeanditblocks
Avoid mocks or stubs in tests; use helpers from@internal/testcontainerswhen Redis or Postgres are needed
Use vitest for unit tests
Files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
🧠 Learnings (3)
📚 Learning: 2025-11-27T16:27:48.109Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: AGENTS.md:0-0
Timestamp: 2025-11-27T16:27:48.109Z
Learning: Applies to **/*.test.{ts,tsx,js,jsx} : Avoid mocks or stubs in tests; use helpers from `internal/testcontainers` when Redis or Postgres are needed
Applied to files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
📚 Learning: 2025-11-14T16:03:06.917Z
Learnt from: matt-aitken
Repo: triggerdotdev/trigger.dev PR: 2681
File: apps/webapp/app/services/platform.v3.server.ts:258-302
Timestamp: 2025-11-14T16:03:06.917Z
Learning: In `apps/webapp/app/services/platform.v3.server.ts`, the `getDefaultEnvironmentConcurrencyLimit` function intentionally throws an error (rather than falling back to org.maximumConcurrencyLimit) when the billing client returns undefined plan limits. This fail-fast behavior prevents users from receiving more concurrency than their plan entitles them to. The org.maximumConcurrencyLimit fallback is only for self-hosted deployments where no billing client exists.
Applied to files:
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 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 (ubuntu-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: typecheck / typecheck
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (1)
packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts (1)
22-210: LGTM! Well-structured test helper for two-stage processing.The
TestFairQueueHelpereffectively orchestratesFairQueueandWorkerQueueManagerfor testing. The consumer loop properly handles message key parsing, retrieves stored messages, and delegates lifecycle operations (heartbeat, complete, release, fail) to the underlyingFairQueue.
…s from the worker queue
PR Review: fix(batch): optimize processing batch trigger v2I've reviewed this pull request which introduces significant architectural changes to the batch queue processing system. Here's my detailed feedback: 📋 SummaryThis PR refactors the FairQueue system from a direct-processing model to a two-stage processing architecture where:
Key changes include:
✅ Strengths
|
PR Review: fix(batch): optimize processing batch trigger v2SummaryThis PR significantly refactors the batch queue and FairQueue architecture to improve performance through a two-stage processing model. The changes introduce horizontal scaling capabilities via sharding and worker queues. Key Architectural Changes
✅ StrengthsCode Quality
Performance
Test Coverage
|
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
packages/redis-worker/src/fair-queue/workerQueue.ts (1)
114-120: Consider simplifying the redundant condition check.Since
cleanupis only non-null whensignalis truthy (line 106), the conditionif (signal && cleanup)could be simplified toif (cleanup). The same applies to line 153.♻️ Proposed simplification
- if (signal && cleanup) { + if (cleanup) { signal.addEventListener("abort", cleanup, { once: true }); if (signal.aborted) {And in the finally block:
- if (cleanup && signal) { + if (cleanup) { signal.removeEventListener("abort", cleanup); }
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
packages/redis-worker/src/fair-queue/workerQueue.ts
🧰 Additional context used
📓 Path-based instructions (3)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
packages/redis-worker/src/fair-queue/workerQueue.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
packages/redis-worker/src/fair-queue/workerQueue.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
packages/redis-worker/src/fair-queue/workerQueue.ts
🧬 Code graph analysis (1)
packages/redis-worker/src/fair-queue/workerQueue.ts (3)
packages/redis-worker/src/fair-queue/telemetry.ts (1)
cleanup(705-708)internal-packages/run-engine/src/batch-queue/completionTracker.ts (1)
cleanup(295-308)internal-packages/run-engine/src/run-queue/index.ts (1)
blockingClient(1962-1966)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 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 (6, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: Cursor Bugbot
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (2)
packages/redis-worker/src/fair-queue/workerQueue.ts (2)
104-110: LGTM! Proper cleanup function hoisting.The cleanup function is correctly hoisted outside the try block, making it accessible in the finally block. This is essential for preventing listener accumulation.
151-159: Excellent fix for the memory leak!The finally block correctly removes the abort listener, preventing accumulation on the AbortSignal. The comment clearly explains why manual cleanup is necessary despite
{ once: true }. The same pattern appears infair-queue/index.tsand already has proper cleanup in place, so no similar issues remain in the codebase.
PR Review: fix(batch): optimize processing batch trigger v2SummaryThis PR significantly refactors the Code Quality and Best Practices ✅Strengths:
Minor suggestions:
Potential Bugs or Issues
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In @internal-packages/run-engine/src/batch-queue/index.ts:
- Around line 703-704: The queue time calculation using Date.now() -
storedMessage.timestamp is measuring time from batch creation
(storedMessage.timestamp is set from meta.createdAt + itemIndex) not the
per-item enqueue moment, which makes itemQueueTimeHistogram.record inaccurate
for streaming batches; to fix either (a) record and store a real per-item
enqueue timestamp when enqueueBatchItem is called (e.g., attach enqueueAt to
each storedMessage item and use that for queue time calculation before calling
itemQueueTimeHistogram.record), or (b) if you intentionally want batch-creation
latency, rename/document the metric to reflect “time since batch creation” and
keep using storedMessage.timestamp; update references to
storedMessage.timestamp, enqueueBatchItem, and itemQueueTimeHistogram.record
accordingly.
- Around line 683-688: When storedMessage is null after calling
this.fairQueue.getMessageData(messageId, queueId) the code currently logs and
returns without cleaning up visibility timeouts; change the branch so that after
logging the error (logger.error(...)) you call await
this.fairQueue.completeMessage(messageId, queueId) before returning. This
mirrors the other error paths and ensures visibilityManager.complete runs even
when storedMessage is missing; update the block around storedMessage, keeping
the log message and adding the await this.fairQueue.completeMessage call and
then return.
🧹 Nitpick comments (1)
internal-packages/run-engine/src/batch-queue/index.ts (1)
634-640: Consider more robust message key parsing.The
indexOfapproach works but could be more explicit about expected format. If message keys could contain multiple colons, this would split at the first colon which may or may not be correct.♻️ Alternative parsing approach
- const colonIndex = messageKey.indexOf(":"); - if (colonIndex === -1) { - this.logger.error("Invalid message key format", { messageKey }); - this.batchedSpanManager.incrementStat(loopId, "invalid_message_keys"); - return false; - } - - const messageId = messageKey.substring(0, colonIndex); - const queueId = messageKey.substring(colonIndex + 1); + const parts = messageKey.split(":", 2); + if (parts.length !== 2) { + this.logger.error("Invalid message key format, expected 'messageId:queueId'", { + messageKey, + expected: "messageId:queueId" + }); + this.batchedSpanManager.incrementStat(loopId, "invalid_message_keys"); + return false; + } + + const [messageId, queueId] = parts;
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
internal-packages/run-engine/src/batch-queue/index.ts
🧰 Additional context used
📓 Path-based instructions (3)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
internal-packages/run-engine/src/batch-queue/index.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
internal-packages/run-engine/src/batch-queue/index.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
internal-packages/run-engine/src/batch-queue/index.ts
🧠 Learnings (4)
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `runs.subscribeToBatch()` to subscribe to changes for all runs in a batch
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-08-14T18:35:44.370Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: apps/webapp/app/env.server.ts:764-765
Timestamp: 2025-08-14T18:35:44.370Z
Learning: The BoolEnv helper in apps/webapp/app/utils/boolEnv.ts uses z.preprocess with inconsistent default value types across the codebase - some usages pass boolean defaults (correct) while others pass string defaults (incorrect), leading to type confusion. The helper should enforce boolean-only defaults or have clearer documentation.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:26:58.661Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-11-27T16:26:58.661Z
Learning: Use `trigger.dev/redis-worker` for background job and worker system needs in the webapp and run engine
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 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: typecheck / typecheck
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (5)
internal-packages/run-engine/src/batch-queue/index.ts (5)
717-722: Defensive message completion is correct.Completing the message even when
processItemCallbackis not set prevents queue blocking. This is the right approach for defensive error handling.
504-514: Past review comment has been addressed.The missing
batchedSpanManager.cleanupAll()call flagged in a previous review has been added at line 508. This ensures proper cleanup of tracing spans during shutdown.
54-202: Well-structured worker-queue architecture.The introduction of a single shared
BATCH_WORKER_QUEUE_IDwithWorkerQueueManagerandBatchedSpanManagerprovides a clean two-stage consumption model. FairQueue handles fair scheduling across environments, while BatchQueue's consumer loops process from the worker queue with proper tracing and lifecycle management.The initialization sequence is correct:
- FairQueue for scheduling and routing
- WorkerQueueManager for worker queue consumption
- BatchedSpanManager for per-loop tracing
455-478: Lifecycle management is idempotent and well-coordinated.The
start()method correctly:
- Guards against double-start with
isRunningcheck- Creates fresh
AbortController- Starts FairQueue's master consumers
- Spawns worker queue consumer loops
This follows proper initialization order and enables clean shutdown via the abort signal.
863-868: Idempotency protection is in place for tracking, but completeMessage coordination remains a concern.The
recordSuccessandrecordFailuremethods use atomic Lua scripts withitemIndexas the idempotency key to prevent double-counting on redelivery. A SADD operation only increments the processed counter if the item hasn't been recorded before, ensuring idempotent behavior.However,
completeMessage(which finalizes the message in FairQueue after recording success/failure) does not employ explicit atomic coordination. IfcompleteMessagefails after the counter has been incremented, the item will be counted as processed but the message won't be marked complete in FairQueue, risking redelivery. While the counter increment is protected from double-counting, the lack of atomic coordination between recording and completion creates a window where the batch could be redelivered with an inconsistent state.
PR Review: fix(batch): optimize processing batch trigger v2This PR implements significant improvements to the BatchQueue system by adding a full two-phase dequeue process in FairQueue and moving worker queue consumption responsibility to BatchQueue. Overall, this is a well-designed architectural improvement with good test coverage. ✅ StrengthsArchitecture & Design:
Configuration Flexibility:
Test Coverage:
Code Quality:
|
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
internal-packages/run-engine/src/batch-queue/index.ts (1)
635-640: Consider adding metrics for invalid message keys.When a message key has an invalid format, the code logs an error and returns false, effectively dropping the message. While this is appropriate (the message was already atomically removed by
blockingPop), consider adding a counter metric to track these occurrences for monitoring and alerting.📊 Optional metric addition
const colonIndex = messageKey.indexOf(":"); if (colonIndex === -1) { this.logger.error("Invalid message key format", { messageKey }); this.batchedSpanManager.incrementStat(loopId, "invalid_message_keys"); + // Consider adding: this.someErrorCounter?.add(1, { error_type: "invalid_message_key" }); return false; }
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
internal-packages/run-engine/src/batch-queue/index.ts
🧰 Additional context used
📓 Path-based instructions (3)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
internal-packages/run-engine/src/batch-queue/index.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
internal-packages/run-engine/src/batch-queue/index.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
internal-packages/run-engine/src/batch-queue/index.ts
🧠 Learnings (4)
📚 Learning: 2025-08-14T18:35:44.370Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: apps/webapp/app/env.server.ts:764-765
Timestamp: 2025-08-14T18:35:44.370Z
Learning: The BoolEnv helper in apps/webapp/app/utils/boolEnv.ts uses z.preprocess with inconsistent default value types across the codebase - some usages pass boolean defaults (correct) while others pass string defaults (incorrect), leading to type confusion. The helper should enforce boolean-only defaults or have clearer documentation.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-08-14T10:09:02.528Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: internal-packages/run-engine/src/engine/index.ts:466-467
Timestamp: 2025-08-14T10:09:02.528Z
Learning: In the triggerdotdev/trigger.dev codebase, it's acceptable to pass `string | undefined` types directly to Prisma operations (both create and update). The codebase consistently uses this pattern and the team is comfortable with how Prisma handles undefined values.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:26:58.661Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-11-27T16:26:58.661Z
Learning: Use `trigger.dev/redis-worker` for background job and worker system needs in the webapp and run engine
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (3)
internal-packages/run-engine/src/batch-queue/index.ts (3)
54-57: LGTM: Clear architecture documentation.The constant and its accompanying comments clearly explain the single shared worker queue design where FairQueue handles fair scheduling before routing all messages to this queue for BatchQueue's consumer loops.
505-514: LGTM: Comprehensive resource cleanup.The close method properly sequences shutdown: stops consumers first, cleans up batched spans, then closes all Redis connections. The addition of
batchedSpanManager.cleanupAll()at line 508 addresses the previous review concern about potential span leaks.
681-689: LGTM: Proper cleanup for missing messages.The code correctly calls
completeMessageeven when stored message data is not found (line 687), ensuring visibility timeout cleanup. This addresses the previous review concern about potential resource leaks.
PR Review: fix(batch): optimize processing batch trigger v2Thanks for submitting this significant refactor of the BatchQueue system. This is a well-structured PR that introduces important architectural improvements. Here's my detailed review: 🎯 SummaryThis PR refactors the FairQueue to implement a full two-phase dequeue architecture where:
This is a solid separation of concerns that should improve scalability and flexibility. ✅ Code Quality & Best PracticesStrengths:
Minor Suggestions:
🐛 Potential Bugs or Issues
⚡ Performance ConsiderationsImprovements:
Considerations:
🔒 Security ConcernsNo security issues identified. The changes are internal queue processing logic with no user input handling or authentication changes. 🧪 Test CoverageStrong test coverage:
Suggestions:
📝 Additional Notes
🏁 OverallThis is a well-executed refactor that improves the architecture while maintaining backward compatibility in the API surface (via the optional new env vars). The test coverage is thorough, and the code quality is high. Recommendation: Approve with minor suggestions Consider addressing the performance considerations around capacity checking frequency if you expect high tenant counts, but the current implementation should work well for typical use cases. |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In @packages/redis-worker/src/fair-queue/index.ts:
- Around line 986-987: The bug is passing message.payload to workerQueueResolver
instead of the full StoredMessage; update the call so workerQueueResolver
receives the complete StoredMessage (same shape constructed in the enqueue
paths) so it can access messageId/queueId/tenantId/etc. Locate the resolver
invocation (workerQueueResolver(...)) and replace the argument with the full
message object (not message.payload), matching the StoredMessage used by the
enqueue methods.
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
internal-packages/run-engine/src/batch-queue/index.tspackages/redis-worker/src/fair-queue/index.tspackages/redis-worker/src/index.tspackages/redis-worker/src/utils.ts
🧰 Additional context used
📓 Path-based instructions (3)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{ts,tsx}: Use types over interfaces for TypeScript
Avoid using enums; prefer string unions or const objects instead
Files:
packages/redis-worker/src/utils.tspackages/redis-worker/src/index.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
**/*.{ts,tsx,js,jsx}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Use function declarations instead of default exports
Files:
packages/redis-worker/src/utils.tspackages/redis-worker/src/index.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
**/*.{js,ts,jsx,tsx,json,md,css,scss}
📄 CodeRabbit inference engine (AGENTS.md)
Format code using Prettier
Files:
packages/redis-worker/src/utils.tspackages/redis-worker/src/index.tspackages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
🧠 Learnings (7)
📚 Learning: 2024-10-18T15:41:52.352Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 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.
Applied to files:
packages/redis-worker/src/utils.tsinternal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Control concurrency using the `queue` property with `concurrencyLimit` option
Applied to files:
packages/redis-worker/src/fair-queue/index.tsinternal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:27:35.304Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/writing-tasks.mdc:0-0
Timestamp: 2025-11-27T16:27:35.304Z
Learning: Applies to **/trigger/**/*.{ts,tsx,js,jsx} : Use `runs.subscribeToBatch()` to subscribe to changes for all runs in a batch
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-08-14T18:35:44.370Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 2390
File: apps/webapp/app/env.server.ts:764-765
Timestamp: 2025-08-14T18:35:44.370Z
Learning: The BoolEnv helper in apps/webapp/app/utils/boolEnv.ts uses z.preprocess with inconsistent default value types across the codebase - some usages pass boolean defaults (correct) while others pass string defaults (incorrect), leading to type confusion. The helper should enforce boolean-only defaults or have clearer documentation.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2024-10-22T10:50:41.311Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 1424
File: packages/core/src/v3/errors.ts:155-189
Timestamp: 2024-10-22T10:50:41.311Z
Learning: When using `assertExhaustive` in a `switch` statement in TypeScript (e.g., in the `shouldRetryError` function in `packages/core/src/v3/errors.ts`), and it throws an error, it's acceptable not to add a `return` statement afterward, as control flow will not proceed beyond the `assertExhaustive` call.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2024-10-07T10:32:30.100Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 1387
File: packages/cli-v3/src/executions/taskRunProcess.ts:408-413
Timestamp: 2024-10-07T10:32:30.100Z
Learning: In the `parseExecuteError` method in `packages/cli-v3/src/executions/taskRunProcess.ts`, using `String(error)` to populate the `message` field works fine and even prepends `error.name`.
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
📚 Learning: 2025-11-27T16:26:58.661Z
Learnt from: CR
Repo: triggerdotdev/trigger.dev PR: 0
File: .cursor/rules/webapp.mdc:0-0
Timestamp: 2025-11-27T16:26:58.661Z
Learning: Use `trigger.dev/redis-worker` for background job and worker system needs in the webapp and run engine
Applied to files:
internal-packages/run-engine/src/batch-queue/index.ts
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (25)
- GitHub Check: Cursor Bugbot
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 8)
- GitHub Check: typecheck / typecheck
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- 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: claude-review
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (11)
packages/redis-worker/src/utils.ts (1)
1-12: LGTM! Well-designed utility for consistent abort error detection.This helper correctly handles both native Node.js
AbortError(which sets.name) and custom abort errors created withnew Error("AbortError")(which sets.message). The documentation clearly explains the dual-check rationale.packages/redis-worker/src/index.ts (1)
3-3: LGTM!Clean addition following the existing barrel export pattern.
packages/redis-worker/src/fair-queue/index.ts (4)
758-770: Good fix for the event listener memory leak.The explicit
removeEventListenercall on timeout (line 766) correctly addresses the accumulation issue. The{ once: true }option only removes the listener when it fires on abort, so this manual cleanup is necessary when the timeout resolves first.
972-983: Good fix for stranded claimed messages.When concurrency reservation fails,
releaseBatchnow correctly releases all remaining messages in the batch (from indexionward) back to the queue. This prevents messages from being stranded in the in-flight set until visibility timeout expires.
773-777: LGTM!Using the new
isAbortErrorutility for consistent abort detection.
1053-1099: LGTM on completeMessage implementation.The method correctly handles the case where stored message data is missing, builds a fallback descriptor, and properly sequences the cleanup operations (visibility completion, concurrency release, master queue update).
internal-packages/run-engine/src/batch-queue/index.ts (5)
505-515: Good addition ofbatchedSpanManager.cleanupAll()inclose().This provides a safety net for cleaning up any remaining batched spans that weren't cleaned up by individual consumer loop
finallyblocks.
686-690: Good fix:completeMessageis now called when stored message is not found.This ensures proper cleanup of visibility timeout entries even when the message data is missing (e.g., due to expiration or race conditions).
666-671: Good fix: UsingisAbortErrorhelper for consistent abort detection.This correctly handles both native
AbortError(which sets.name) and custom abort errors (which set.message), consistent with the FairQueue implementation.
604-676: LGTM on worker queue consumer loop implementation.Clean loop structure with proper:
- Batched span management (init, per-iteration spans, cleanup in finally)
- Abort signal handling at multiple levels
- Message key parsing with error handling
- Statistics tracking for observability
55-58: Clear documentation of the single worker queue design.Good explanation of why BatchQueue uses a single shared worker queue - FairQueue handles fair scheduling at the master queue level, then routes all messages to this queue for BatchQueue's own consumer loop to process.
This PR fixes some issues with the new BatchQueue by implementing the full two-phase dequeue process in the FairQueue, and moving the responsibility of consuming the worker queue to the BatchQueue and independently enabling it via the
BATCH_QUEUE_WORKER_QUEUE_ENABLEDenv var. We've also introduced theBATCH_QUEUE_SHARD_COUNTenv var to control the count of master queue shards in the FairQueue. We can also control how many queues are considered in each iteration of the master queue consumer via theBATCH_QUEUE_MASTER_QUEUE_LIMITenv var.This PR will also now skip trying to dequeue from tenants that are at concurrency capacity, which should lead to fewer issues with low concurrency tenants blocking higher concurrency tenants from processing.