Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 006951a

Browse filesBrowse files
ericallamericallam
authored
Set a maximum number of times the dequeue loop can run (triggerdotdev#1950)
1 parent 0ebecbe commit 006951a
Copy full SHA for 006951a

6 files changed

+11-106Lines changed: 11 additions & 106 deletions

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎apps/webapp/app/env.server.ts‎

Copy file name to clipboardExpand all lines: apps/webapp/app/env.server.ts
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -459,6 +459,7 @@ const EnvironmentSchema = z.object({
459459
RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
460460
RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
461461
RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
462+
RUN_ENGINE_MAX_DEQUEUE_LOOP_ATTEMPTS: z.coerce.number().int().default(10),
462463

463464
RUN_ENGINE_WORKER_REDIS_HOST: z
464465
.string()
Collapse file

‎apps/webapp/app/v3/runEngine.server.ts‎

Copy file name to clipboardExpand all lines: apps/webapp/app/v3/runEngine.server.ts
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ function createRunEngine() {
5858
maximumEnvCount: env.RUN_ENGINE_MAXIMUM_ENV_COUNT,
5959
tracer,
6060
},
61+
maxDequeueLoopAttempts: env.RUN_ENGINE_MAX_DEQUEUE_LOOP_ATTEMPTS,
6162
},
6263
runLock: {
6364
redis: {
Collapse file

‎internal-packages/run-engine/src/engine/index.ts‎

Copy file name to clipboardExpand all lines: internal-packages/run-engine/src/engine/index.ts
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ export class RunEngine {
109109
logger: new Logger("RunQueue", "debug"),
110110
redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
111111
retryOptions: options.queue?.retryOptions,
112+
maxDequeueLoopAttempts: options.queue?.maxDequeueLoopAttempts ?? 10,
112113
});
113114

114115
this.worker = new Worker({
Collapse file

‎internal-packages/run-engine/src/engine/types.ts‎

Copy file name to clipboardExpand all lines: internal-packages/run-engine/src/engine/types.ts
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ export type RunEngineOptions = {
2929
FairQueueSelectionStrategyOptions,
3030
"parentQueueLimit" | "tracer" | "biases" | "reuseSnapshotCount" | "maximumEnvCount"
3131
>;
32+
maxDequeueLoopAttempts?: number;
3233
};
3334
runLock: {
3435
redis: RedisOptions;
Collapse file

‎internal-packages/run-engine/src/run-queue/index.ts‎

Copy file name to clipboardExpand all lines: internal-packages/run-engine/src/run-queue/index.ts
+7-15Lines changed: 7 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ export type RunQueueOptions = {
5151
verbose?: boolean;
5252
logger?: Logger;
5353
retryOptions?: RetryOptions;
54+
maxDequeueLoopAttempts?: number;
5455
};
5556

5657
type DequeuedMessage = {
@@ -77,6 +78,7 @@ export class RunQueue {
7778
private redis: Redis;
7879
public keys: RunQueueKeyProducer;
7980
private queueSelectionStrategy: RunQueueSelectionStrategy;
81+
private maxDequeueLoopAttempts: number;
8082

8183
constructor(private readonly options: RunQueueOptions) {
8284
this.retryOptions = options.retryOptions ?? defaultRetrySettings;
@@ -92,6 +94,7 @@ export class RunQueue {
9294

9395
this.keys = options.keys;
9496
this.queueSelectionStrategy = options.queueSelectionStrategy;
97+
this.maxDequeueLoopAttempts = options.maxDequeueLoopAttempts ?? 10;
9598

9699
this.subscriber = createRedisClient(options.redis, {
97100
onError: (error) => {
@@ -393,6 +396,7 @@ export class RunQueue {
393396

394397
let attemptedEnvs = 0;
395398
let attemptedQueues = 0;
399+
let dequeueLoopAttempts = 0;
396400

397401
const messages: DequeuedMessage[] = [];
398402

@@ -404,16 +408,13 @@ export class RunQueue {
404408
tenantQueues[env.envId] = [...env.queues]; // Create a copy of the queues array
405409
}
406410

407-
// Track if we successfully dequeued any message in a complete cycle
408-
let successfulDequeueInCycle = false;
409-
410411
// Continue until we've hit max count or all tenants have empty queue lists
411412
while (
412413
messages.length < maxCount &&
413-
Object.values(tenantQueues).some((queues) => queues.length > 0)
414+
Object.values(tenantQueues).some((queues) => queues.length > 0) &&
415+
dequeueLoopAttempts < this.maxDequeueLoopAttempts
414416
) {
415-
// Reset the success flag at the start of each cycle
416-
successfulDequeueInCycle = false;
417+
dequeueLoopAttempts++;
417418

418419
for (const env of envQueues) {
419420
attemptedEnvs++;
@@ -434,7 +435,6 @@ export class RunQueue {
434435

435436
if (message) {
436437
messages.push(message);
437-
successfulDequeueInCycle = true;
438438
// Re-add this queue at the end, since it might have more messages
439439
tenantQueues[env.envId].push(queue);
440440
}
@@ -445,14 +445,6 @@ export class RunQueue {
445445
break;
446446
}
447447
}
448-
449-
// If we completed a full cycle through all tenants with no successful dequeues,
450-
// exit early as we're likely hitting concurrency limits or have no ready messages
451-
if (!successfulDequeueInCycle) {
452-
// IMPORTANT: Keep this log message as it's used in tests
453-
this.logger.log("No successful dequeues in a full cycle, exiting...");
454-
break;
455-
}
456448
}
457449

458450
span.setAttributes({
Collapse file

‎internal-packages/run-engine/src/run-queue/tests/dequeueMessageFromMasterQueue.test.ts‎

Copy file name to clipboardExpand all lines: internal-packages/run-engine/src/run-queue/tests/dequeueMessageFromMasterQueue.test.ts
-91Lines changed: 0 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -262,95 +262,4 @@ describe("RunQueue.dequeueMessageFromMasterQueue", () => {
262262
}
263263
}
264264
);
265-
266-
redisTest(
267-
"should exit early when no messages can be dequeued in a full cycle",
268-
async ({ redisContainer }) => {
269-
const mockLogger = {
270-
log: vi.fn(),
271-
error: vi.fn(),
272-
warn: vi.fn(),
273-
debug: vi.fn(),
274-
name: "test-logger",
275-
level: "debug",
276-
filteredKeys: [],
277-
additionalFields: {},
278-
setLevel: vi.fn(),
279-
setFilteredKeys: vi.fn(),
280-
setAdditionalFields: vi.fn(),
281-
child: vi.fn(),
282-
};
283-
284-
const queue = new RunQueue({
285-
...testOptions,
286-
queueSelectionStrategy: new FairQueueSelectionStrategy({
287-
redis: {
288-
keyPrefix: "runqueue:test:",
289-
host: redisContainer.getHost(),
290-
port: redisContainer.getPort(),
291-
},
292-
keys: testOptions.keys,
293-
}),
294-
redis: {
295-
keyPrefix: "runqueue:test:",
296-
host: redisContainer.getHost(),
297-
port: redisContainer.getPort(),
298-
},
299-
// @ts-expect-error
300-
logger: mockLogger,
301-
});
302-
303-
try {
304-
const envMasterQueue = `env:${authenticatedEnvDev.id}`;
305-
const queueCount = 10; // Reduced for simplicity
306-
307-
// First, create all queues and enqueue initial messages
308-
for (let i = 0; i < queueCount; i++) {
309-
const queueName = `${messageDev.queue}_${i}`;
310-
// Set each queue's concurrency limit to 0 (this guarantees dequeue will fail)
311-
await queue.updateQueueConcurrencyLimits(authenticatedEnvDev, queueName, 0);
312-
313-
// Enqueue a message to each queue
314-
await queue.enqueueMessage({
315-
env: authenticatedEnvDev,
316-
message: { ...messageDev, runId: `r${4321 + i}`, queue: queueName },
317-
masterQueues: ["main", envMasterQueue],
318-
});
319-
}
320-
321-
// Try to dequeue messages - this should exit early due to concurrency limits
322-
const startTime = Date.now();
323-
const dequeued = await queue.dequeueMessageFromMasterQueue(
324-
"test_12345",
325-
envMasterQueue,
326-
queueCount
327-
);
328-
const endTime = Date.now();
329-
330-
// Verify no messages were dequeued
331-
expect(dequeued.length).toBe(0);
332-
333-
// Verify the operation completed quickly (under 1000ms)
334-
const duration = endTime - startTime;
335-
expect(duration).toBeLessThan(1000);
336-
337-
// Verify we only logged one early exit message
338-
expect(mockLogger.log).toHaveBeenCalledWith(
339-
expect.stringContaining("No successful dequeues in a full cycle, exiting")
340-
);
341-
expect(mockLogger.log.mock.calls.length).toBeLessThanOrEqual(2);
342-
343-
// Verify all messages are still in queues
344-
let totalRemaining = 0;
345-
for (let i = 0; i < queueCount; i++) {
346-
const queueName = `${messageDev.queue}_${i}`;
347-
const length = await queue.lengthOfQueue(authenticatedEnvDev, queueName);
348-
totalRemaining += length;
349-
}
350-
expect(totalRemaining).toBe(queueCount);
351-
} finally {
352-
await queue.quit();
353-
}
354-
}
355-
);
356265
});

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.