Add runs to an env queue, as well as the actual queue

Author: ericallam

apps/webapp/app/v3/marqs/index.server.ts

@@ -1074,6 +1074,7 @@ export class MarQS {
       concurrencyKey,
       envConcurrencyKey,
       orgConcurrencyKey,
+      this.keys.envQueueKeyFromQueue(message.queue),
       message.queue,
       message.messageId,
       JSON.stringify(message),
@@ -1111,6 +1112,7 @@ export class MarQS {
       currentConcurrencyKey,
       envCurrentConcurrencyKey,
       orgCurrentConcurrencyKey,
+      this.keys.envQueueKeyFromQueue(messageQueue),
       messageQueue,
       String(Date.now()),
       String(this.options.defaultEnvConcurrency),
@@ -1187,6 +1189,7 @@ export class MarQS {
       concurrencyKey,
       envConcurrencyKey,
       orgConcurrencyKey,
+      this.keys.envQueueKeyFromQueue(messageQueue),
       messageId,
       messageQueue
     );
@@ -1234,6 +1237,7 @@ export class MarQS {
       envConcurrencyKey,
       orgConcurrencyKey,
       visibilityQueue,
+      this.keys.envQueueKeyFromQueue(messageQueue),
       messageQueue,
       messageId,
       String(Date.now()),
@@ -1347,14 +1351,15 @@ export class MarQS {
 
   #registerCommands() {
     this.redis.defineCommand("enqueueMessage", {
-      numberOfKeys: 6,
+      numberOfKeys: 7,
       lua: `
 local queue = KEYS[1]
 local parentQueue = KEYS[2]
 local messageKey = KEYS[3]
 local concurrencyKey = KEYS[4]
 local envCurrentConcurrencyKey = KEYS[5]
 local orgCurrentConcurrencyKey = KEYS[6]
+local envQueue = KEYS[7]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -1367,6 +1372,9 @@ redis.call('SET', messageKey, messageData)
 -- Add the message to the queue
 redis.call('ZADD', queue, messageScore, messageId)
 
+-- Add the message to the env queue
+redis.call('ZADD', envQueue, messageScore, messageId)
+
 -- Rebalance the parent queue
 local earliestMessage = redis.call('ZRANGE', queue, 0, 0, 'WITHSCORES')
 if #earliestMessage == 0 then
@@ -1383,7 +1391,7 @@ redis.call('SREM', orgCurrentConcurrencyKey, messageId)
     });
 
     this.redis.defineCommand("dequeueMessage", {
-      numberOfKeys: 8,
+      numberOfKeys: 9,
       lua: `
 -- Keys: childQueue, parentQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local childQueue = KEYS[1]
@@ -1394,6 +1402,7 @@ local orgConcurrencyLimitKey = KEYS[5]
 local currentConcurrencyKey = KEYS[6]
 local envCurrentConcurrencyKey = KEYS[7]
 local orgCurrentConcurrencyKey = KEYS[8]
+local envQueueKey = KEYS[9]
 
 -- Args: childQueueName, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
 local childQueueName = ARGV[1]
@@ -1438,6 +1447,7 @@ local messageScore = tonumber(messages[2])
 
 -- Move message to timeout queue and update concurrency
 redis.call('ZREM', childQueue, messageId)
+redis.call('ZREM', envQueueKey, messageId)
 redis.call('SADD', currentConcurrencyKey, messageId)
 redis.call('SADD', envCurrentConcurrencyKey, messageId)
 redis.call('SADD', orgCurrentConcurrencyKey, messageId)
@@ -1474,7 +1484,7 @@ redis.call('SET', messageKey, messageData, 'GET')
     });
 
     this.redis.defineCommand("acknowledgeMessage", {
-      numberOfKeys: 7,
+      numberOfKeys: 8,
       lua: `
 -- Keys: parentQueue, messageKey, messageQueue, visibilityQueue, concurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local parentQueue = KEYS[1]
@@ -1484,6 +1494,7 @@ local visibilityQueue = KEYS[4]
 local concurrencyKey = KEYS[5]
 local envCurrentConcurrencyKey = KEYS[6]
 local orgCurrentConcurrencyKey = KEYS[7]
+local envQueueKey = KEYS[8]
 
 -- Args: messageId, messageQueueName
 local messageId = ARGV[1]
@@ -1495,6 +1506,9 @@ redis.call('DEL', messageKey)
 -- Remove the message from the queue
 redis.call('ZREM', messageQueue, messageId)
 
+-- Remove the message from the env queue
+redis.call('ZREM', envQueueKey, messageId)
+
 -- Rebalance the parent queue
 local earliestMessage = redis.call('ZRANGE', messageQueue, 0, 0, 'WITHSCORES')
 if #earliestMessage == 0 then
@@ -1514,7 +1528,7 @@ redis.call('SREM', orgCurrentConcurrencyKey, messageId)
     });
 
     this.redis.defineCommand("nackMessage", {
-      numberOfKeys: 7,
+      numberOfKeys: 8,
       lua: `
 -- Keys: childQueueKey, parentQueueKey, visibilityQueue, concurrencyKey, envConcurrencyKey, orgConcurrencyKey, messageId
 local messageKey = KEYS[1]
@@ -1524,6 +1538,7 @@ local concurrencyKey = KEYS[4]
 local envConcurrencyKey = KEYS[5]
 local orgConcurrencyKey = KEYS[6]
 local visibilityQueue = KEYS[7]
+local envQueueKey = KEYS[8]
 
 -- Args: childQueueName, messageId, currentTime, messageScore
 local childQueueName = ARGV[1]
@@ -1547,6 +1562,9 @@ end
 -- Enqueue the message into the queue
 redis.call('ZADD', childQueueKey, messageScore, messageId)
 
+-- Enqueue the message into the env queue
+redis.call('ZADD', envQueueKey, messageScore, messageId)
+
 -- Rebalance the parent queue
 local earliestMessage = redis.call('ZRANGE', childQueueKey, 0, 0, 'WITHSCORES')
 if #earliestMessage == 0 then
@@ -1729,6 +1747,7 @@ declare module "ioredis" {
       concurrencyKey: string,
       envConcurrencyKey: string,
       orgConcurrencyKey: string,
+      envQueue: string,
       queueName: string,
       messageId: string,
       messageData: string,
@@ -1745,6 +1764,7 @@ declare module "ioredis" {
       currentConcurrencyKey: string,
       envCurrentConcurrencyKey: string,
       orgCurrentConcurrencyKey: string,
+      envQueueKey: string,
       childQueueName: string,
       currentTime: string,
       defaultEnvConcurrencyLimit: string,
@@ -1766,6 +1786,7 @@ declare module "ioredis" {
       concurrencyKey: string,
       envConcurrencyKey: string,
       orgConcurrencyKey: string,
+      envQueueKey: string,
       messageId: string,
       messageQueueName: string,
       callback?: Callback<void>
@@ -1779,6 +1800,7 @@ declare module "ioredis" {
       envConcurrencyKey: string,
       orgConcurrencyKey: string,
       visibilityQueue: string,
+      envQueueKey: string,
       childQueueName: string,
       messageId: string,
       currentTime: string,

apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts

@@ -128,6 +128,12 @@ export class MarQSShortKeyProducer implements MarQSKeyProducer {
     return [this.envKeySection(env.id), constants.CURRENT_CONCURRENCY_PART].join(":");
   }
 
+  envQueueKeyFromQueue(queue: string) {
+    const envId = this.normalizeQueue(queue).split(":")[3];
+
+    return `${constants.ENV_PART}:${envId}:${constants.QUEUE_PART}`;
+  }
+
   messageKey(messageId: string) {
     return `${constants.MESSAGE_PART}:${messageId}`;
   }

apps/webapp/app/v3/marqs/types.ts

@@ -44,6 +44,7 @@ export interface MarQSKeyProducer {
   envCurrentConcurrencyKeyFromQueue(queue: string): string;
   orgCurrentConcurrencyKey(env: AuthenticatedEnvironment): string;
   envCurrentConcurrencyKey(env: AuthenticatedEnvironment): string;
+  envQueueKeyFromQueue(queue: string): string;
   messageKey(messageId: string): string;
   stripKeyPrefix(key: string): string;
 }

references/v3-catalog/src/trigger/simple.ts

@@ -208,3 +208,10 @@ export const childTask = task({
     };
   },
 });
+
+export const retryTask = task({
+  id: "retry-task",
+  run: async (payload: any) => {
+    throw new Error("This task will always fail");
+  },
+});

references/v3-catalog/trigger.config.ts

@@ -18,7 +18,7 @@ export default defineConfig({
   instrumentations: [new OpenAIInstrumentation()],
   additionalFiles: ["wrangler/wrangler.toml"],
   retries: {
-    enabledInDev: false,
+    enabledInDev: true,
     default: {
       maxAttempts: 10,
       minTimeoutInMs: 5_000,