Skip to content

fix(v3): eagerly dequeue messages from a queue when that queue is added to or removed from (v4 backport) #2438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ericallam merged 2 commits into from
Aug 22, 2025
Merged

fix(v3): eagerly dequeue messages from a queue when that queue is added to or removed from (v4 backport) #2438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
33d134f
fix(v3): eagerly dequeue messages from a queue when that queue is add…
ericallam Aug 22, 2025
d44a525
fixed configuration
ericallam Aug 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions apps/webapp/app/env.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,15 @@ const EnvironmentSchema = z.object({
MARQS_SHARED_WORKER_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(250),
MARQS_SHARED_WORKER_QUEUE_MAX_MESSAGE_COUNT: z.coerce.number().int().default(10),

MARQS_SHARED_WORKER_QUEUE_EAGER_DEQUEUE_ENABLED: z.string().default("0"),
MARQS_WORKER_ENABLED: z.string().default("0"),
MARQS_WORKER_COUNT: z.coerce.number().int().default(2),
MARQS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(5),
MARQS_WORKER_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS: z.coerce.number().int().default(100),
MARQS_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),

PROD_TASK_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),

VERBOSE_GRAPHILE_LOGGING: z.string().default("false"),
Expand Down
152 changes: 152 additions & 0 deletions apps/webapp/app/v3/marqs/index.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import {
trace,
Tracer,
} from "@opentelemetry/api";
import { type RedisOptions } from "@internal/redis";
import {
SEMATTRS_MESSAGE_ID,
SEMATTRS_MESSAGING_SYSTEM,
Expand Down Expand Up @@ -44,6 +45,9 @@ import {
} from "./constants.server";
import { setInterval } from "node:timers/promises";
import { tryCatch } from "@trigger.dev/core/utils";
import { Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
import z from "zod";
import { Logger } from "@trigger.dev/core/logger";

const KEY_PREFIX = "marqs:";

Expand Down Expand Up @@ -74,6 +78,25 @@ export type MarQSOptions = {
subscriber?: MessageQueueSubscriber;
sharedWorkerQueueConsumerIntervalMs?: number;
sharedWorkerQueueMaxMessageCount?: number;
eagerDequeuingEnabled?: boolean;
workerOptions: {
pollIntervalMs?: number;
immediatePollIntervalMs?: number;
shutdownTimeoutMs?: number;
concurrency?: WorkerConcurrencyOptions;
enabled?: boolean;
redisOptions: RedisOptions;
};
};

const workerCatalog = {
processQueueForWorkerQueue: {
schema: z.object({
queueKey: z.string(),
parentQueueKey: z.string(),
}),
visibilityTimeoutMs: 30_000,
},
};

/**
Expand All @@ -83,6 +106,7 @@ export class MarQS {
private redis: Redis;
public keys: MarQSKeyProducer;
#rebalanceWorkers: Array<AsyncWorker> = [];
private worker: Worker<typeof workerCatalog>;

constructor(private readonly options: MarQSOptions) {
this.redis = options.redis;
Expand All @@ -91,6 +115,26 @@ export class MarQS {

this.#startRebalanceWorkers();
this.#registerCommands();

this.worker = new Worker({
name: "marqs-worker",
redisOptions: options.workerOptions.redisOptions,
catalog: workerCatalog,
concurrency: options.workerOptions?.concurrency,
pollIntervalMs: options.workerOptions?.pollIntervalMs ?? 1000,
immediatePollIntervalMs: options.workerOptions?.immediatePollIntervalMs ?? 100,
shutdownTimeoutMs: options.workerOptions?.shutdownTimeoutMs ?? 10_000,
logger: new Logger("MarQSWorker", "info"),
jobs: {
processQueueForWorkerQueue: async (job) => {
await this.#processQueueForWorkerQueue(job.payload.queueKey, job.payload.parentQueueKey);
},
},
});

if (options.workerOptions?.enabled) {
this.worker.start();
}
}

get name() {
Expand Down Expand Up @@ -280,6 +324,21 @@ export class MarQS {
span.setAttribute("reserve_recursive_queue", reserve.recursiveQueue);
}

if (env.type !== "DEVELOPMENT" && this.options.eagerDequeuingEnabled) {
// This will move the message to the worker queue so it can be dequeued
await this.worker.enqueueOnce({
id: messageQueue, // dedupe by environment, queue, and concurrency key
job: "processQueueForWorkerQueue",
payload: {
queueKey: messageQueue,
parentQueueKey: parentQueue,
},
// Add a small delay to dedupe messages so at most one of these will processed,
// every 500ms per queue, concurrency key, and environment
availableAt: new Date(Date.now() + 500), // 500ms from now
});
}
Comment on lines +327 to +340
Copy link
Contributor

@coderabbitai coderabbitai bot Aug 22, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Gate eager-dequeue behind worker enabled and DRY the 500ms magic number.

If the worker is disabled, enqueueOnce will accumulate jobs that never run. Also, replace the hardcoded 500ms with a shared constant.

Apply:

-        if (env.type !== "DEVELOPMENT" && this.options.eagerDequeuingEnabled) {
+        if (
+          env.type !== "DEVELOPMENT" &&
+          this.options.eagerDequeuingEnabled &&
+          this.options.workerOptions?.enabled
+        ) {
           // This will move the message to the worker queue so it can be dequeued
           await this.worker.enqueueOnce({
             id: messageQueue, // dedupe by environment, queue, and concurrency key
             job: "processQueueForWorkerQueue",
             payload: {
               queueKey: messageQueue,
               parentQueueKey: parentQueue,
             },
             // Add a small delay to dedupe messages so at most one of these will processed,
             // every 500ms per queue, concurrency key, and environment
-            availableAt: new Date(Date.now() + 500), // 500ms from now
+            availableAt: new Date(Date.now() + EAGER_DEQUEUE_DEBOUNCE_MS),
           });
         }

Add once near KEY_PREFIX:

const EAGER_DEQUEUE_DEBOUNCE_MS = 500;
🤖 Prompt for AI Agents 
In apps/webapp/app/v3/marqs/index.server.ts around lines 327-340, the
eager-dequeue enqueueOnce call is executed even when the worker is disabled and
uses a hardcoded 500ms; add a constant EAGER_DEQUEUE_DEBOUNCE_MS = 500 near the
existing KEY_PREFIX declaration and replace the inline 500 with that constant,
and expand the condition to require the worker to be enabled (e.g. check
this.worker?.enabled or this.options.workerEnabled depending on the codebase) so
the enqueueOnce is only called when env.type !== "DEVELOPMENT",
eagerDequeuingEnabled is true, and the worker is enabled.


const result = await this.#callEnqueueMessage(messagePayload, reserve);

if (result) {
Expand Down Expand Up @@ -870,6 +929,64 @@ export class MarQS {
);
}

async #processQueueForWorkerQueue(queueKey: string, parentQueueKey: string) {
return this.#trace("processQueueForWorkerQueue", async (span) => {
span.setAttributes({
[SemanticAttributes.QUEUE]: queueKey,
[SemanticAttributes.PARENT_QUEUE]: parentQueueKey,
});

const maxCount = this.options.sharedWorkerQueueMaxMessageCount ?? 10;

const dequeuedMessages = await this.#callDequeueMessages({
messageQueue: queueKey,
parentQueue: parentQueueKey,
maxCount,
});

if (!dequeuedMessages || dequeuedMessages.length === 0) {
return;
}

await this.#trace(
"addToWorkerQueue",
async (addToWorkerQueueSpan) => {
const workerQueueKey = this.keys.sharedWorkerQueueKey();

addToWorkerQueueSpan.setAttributes({
message_count: dequeuedMessages.length,
[SemanticAttributes.PARENT_QUEUE]: workerQueueKey,
});

await this.redis.rpush(
workerQueueKey,
...dequeuedMessages.map((message) => message.messageId)
);
},
{
kind: SpanKind.INTERNAL,
attributes: {
[SEMATTRS_MESSAGING_OPERATION]: "receive",
[SEMATTRS_MESSAGING_SYSTEM]: "marqs",
},
}
);

// If we dequeued the max count, we need to enqueue another job to dequeue the next batch
if (dequeuedMessages.length === maxCount) {
await this.worker.enqueueOnce({
id: queueKey,
job: "processQueueForWorkerQueue",
payload: {
queueKey,
parentQueueKey,
},
availableAt: new Date(Date.now() + 500), // 500ms from now
});
}
});
}

public async acknowledgeMessage(messageId: string, reason: string = "unknown") {
return this.#trace(
"acknowledgeMessage",
Expand Down Expand Up @@ -901,6 +1018,20 @@ export class MarQS {
messageId,
});

const sharedQueueKey = this.keys.sharedQueueKey();

if (this.options.eagerDequeuingEnabled && message.parentQueue === sharedQueueKey) {
await this.worker.enqueueOnce({
id: message.queue,
job: "processQueueForWorkerQueue",
payload: {
queueKey: message.queue,
parentQueueKey: message.parentQueue,
},
availableAt: new Date(Date.now() + 500), // 500ms from now
});
}

await this.options.subscriber?.messageAcked(message);
},
{
Expand Down Expand Up @@ -2482,5 +2613,26 @@ function getMarQSClient() {
subscriber: concurrencyTracker,
sharedWorkerQueueConsumerIntervalMs: env.MARQS_SHARED_WORKER_QUEUE_CONSUMER_INTERVAL_MS,
sharedWorkerQueueMaxMessageCount: env.MARQS_SHARED_WORKER_QUEUE_MAX_MESSAGE_COUNT,
eagerDequeuingEnabled: env.MARQS_SHARED_WORKER_QUEUE_EAGER_DEQUEUE_ENABLED === "1",
workerOptions: {
enabled: env.MARQS_WORKER_ENABLED === "1",
pollIntervalMs: env.MARQS_WORKER_POLL_INTERVAL_MS,
immediatePollIntervalMs: env.MARQS_WORKER_IMMEDIATE_POLL_INTERVAL_MS,
shutdownTimeoutMs: env.MARQS_WORKER_SHUTDOWN_TIMEOUT_MS,
concurrency: {
workers: env.MARQS_WORKER_COUNT,
tasksPerWorker: env.MARQS_WORKER_CONCURRENCY_TASKS_PER_WORKER,
limit: env.MARQS_WORKER_CONCURRENCY_LIMIT,
},
redisOptions: {
keyPrefix: KEY_PREFIX,
port: env.REDIS_PORT ?? undefined,
host: env.REDIS_HOST ?? undefined,
username: env.REDIS_USERNAME ?? undefined,
password: env.REDIS_PASSWORD ?? undefined,
enableAutoPipelining: true,
...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
},
},
});
}