fix(events): fix ordering — update all Lua scripts for global concurrency release

Previous commit only updated dequeue/release/enqueue Lua scripts but missed:
- acknowledgeMessage: runs completing weren't releasing global concurrency
- nackMessage: nacked runs weren't releasing global concurrency
- moveToDeadLetterQueue: DLQ'd runs weren't releasing
- clearMessageFromConcurrencySets: cleanup wasn't releasing

Also: set globalConcurrencyLimit on subscriber task's queue (not dedicated queue)
since the dev worker only monitors task queues, not custom queues.

Removed queue override from PublishEventService — runs stay in the task's
own queue and ordering is enforced by concurrencyKey + concurrencyLimit:1
+ globalConcurrencyLimit:N.

E2E verified: globalConcurrencyLimit=2 with 3 keys correctly limits to max
2 concurrent runs while maintaining per-key ordering.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: giovaborgogno

apps/webapp/app/v3/services/createBackgroundWorker.server.ts

@@ -373,7 +373,7 @@ async function syncWorkerEvents(
           orderingQueueName,
           1, // per-key limit: always 1 for strict ordering
           orderingQueueName,
-          "SHARED",
+          "NAMED",
           worker,
           prisma
         );
@@ -426,6 +426,20 @@ async function syncWorkerEvents(
       }
     }
 
+    // Look up event ordering config for this task's event
+    const eventManifest = metadata.events?.find((e) => e.id === task.onEvent);
+    const eventOrdering = eventManifest?.ordering;
+
+    // If the event has ordering config, set globalConcurrencyLimit on this task's queue
+    if (eventOrdering?.concurrencyLimit) {
+      const taskQueueName = `task/${task.id}`;
+      await updateGlobalQueueConcurrencyLimits(
+        environment,
+        taskQueueName,
+        eventOrdering.concurrencyLimit
+      );
+    }
+
     const subscription = await prisma.eventSubscription.upsert({
       where: {
         eventDefinitionId_taskSlug_environmentId: {

apps/webapp/app/v3/services/events/publishEvent.server.ts

@@ -303,11 +303,6 @@ export class PublishEventService extends BaseService {
                 : undefined,
               metadata: eventMetadata,
               delay: options.delay,
-              // When ordering key is present, route to dedicated ordering queue
-              // with concurrencyLimit:1 per key + global limit
-              queue: options.orderingKey
-                ? { name: `evt-order:${eventSlug}` }
-                : undefined,
               concurrencyKey: options.orderingKey
                 ? `evt:${eventSlug}:${options.orderingKey}`
                 : undefined,

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

@@ -2063,6 +2063,7 @@ export class RunQueue {
       envCurrentDequeuedKey,
       envQueueKey,
       workerQueueKey,
+      this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(message.queue),
       messageId,
       messageQueue,
       messageKeyValue,
@@ -2105,6 +2106,7 @@ export class RunQueue {
       envCurrentConcurrencyKey,
       queueCurrentDequeuedKey,
       envCurrentDequeuedKey,
+      this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(queue),
       messageId
     );
   }
@@ -2151,6 +2153,7 @@ export class RunQueue {
       queueCurrentDequeuedKey,
       envCurrentDequeuedKey,
       envQueueKey,
+      this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(message.queue),
       //args
       messageId,
       messageQueue,
@@ -2184,6 +2187,7 @@ export class RunQueue {
       envCurrentDequeuedKey,
       envQueueKey,
       deadLetterQueueKey,
+      this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(message.queue),
       messageId,
       messageQueue
     );
@@ -2919,7 +2923,7 @@ return message
     });
 
     this.redis.defineCommand("acknowledgeMessage", {
-      numberOfKeys: 9,
+      numberOfKeys: 10,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -2931,6 +2935,7 @@ local queueCurrentDequeuedKey = KEYS[6]
 local envCurrentDequeuedKey = KEYS[7]
 local envQueueKey = KEYS[8]
 local workerQueueKey = KEYS[9]
+local globalCurrentConcurrencyKey = KEYS[10]
 
 -- Args:
 local messageId = ARGV[1]
@@ -2955,6 +2960,7 @@ end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -2967,7 +2973,7 @@ end
     });
 
     this.redis.defineCommand("nackMessage", {
-      numberOfKeys: 8,
+      numberOfKeys: 9,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -2978,6 +2984,7 @@ local envCurrentConcurrencyKey = KEYS[5]
 local queueCurrentDequeuedKey = KEYS[6]
 local envCurrentDequeuedKey = KEYS[7]
 local envQueueKey = KEYS[8]
+local globalCurrentConcurrencyKey = KEYS[9]
 
 -- Args:
 local messageId = ARGV[1]
@@ -2990,6 +2997,7 @@ redis.call('SET', messageKey, messageData)
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -3009,7 +3017,7 @@ end
     });
 
     this.redis.defineCommand("moveToDeadLetterQueue", {
-      numberOfKeys: 9,
+      numberOfKeys: 10,
       lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -3021,6 +3029,7 @@ local queueCurrentDequeuedKey = KEYS[6]
 local envCurrentDequeuedKey = KEYS[7]
 local envQueueKey = KEYS[8]
 local deadLetterQueueKey = KEYS[9]
+local globalCurrentConcurrencyKey = KEYS[10]
 
 -- Args:
 local messageId = ARGV[1]
@@ -3043,6 +3052,7 @@ redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageI
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -3151,19 +3161,21 @@ return results
     });
 
     this.redis.defineCommand("clearMessageFromConcurrencySets", {
-      numberOfKeys: 4,
+      numberOfKeys: 5,
       lua: `
 -- Keys:
 local queueCurrentConcurrencyKey = KEYS[1]
 local envCurrentConcurrencyKey = KEYS[2]
 local queueCurrentDequeuedKey = KEYS[3]
 local envCurrentDequeuedKey = KEYS[4]
+local globalCurrentConcurrencyKey = KEYS[5]
 
 -- Args:
 local messageId = ARGV[1]
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
@@ -3285,6 +3297,7 @@ declare module "@internal/redis" {
       envCurrentDequeuedKey: string,
       envQueueKey: string,
       workerQueueKey: string,
+      globalCurrentConcurrencyKey: string,
       // args
       messageId: string,
       messageQueueName: string,
@@ -3299,6 +3312,7 @@ declare module "@internal/redis" {
       envCurrentConcurrencyKey: string,
       queueCurrentDequeuedKey: string,
       envCurrentDequeuedKey: string,
+      globalCurrentConcurrencyKey: string,
       // args
       messageId: string,
       callback?: Callback<void>
@@ -3314,6 +3328,7 @@ declare module "@internal/redis" {
       queueCurrentDequeuedKey: string,
       envCurrentDequeuedKey: string,
       envQueueKey: string,
+      globalCurrentConcurrencyKey: string,
       // args
       messageId: string,
       messageQueueName: string,
@@ -3333,6 +3348,7 @@ declare module "@internal/redis" {
       envCurrentDequeuedKey: string,
       envQueueKey: string,
       deadLetterQueueKey: string,
+      globalCurrentConcurrencyKey: string,
       // args
       messageId: string,
       messageQueueName: string,

references/hello-world/src/trigger/events-test.ts

@@ -0,0 +1,29 @@
+import { event, task, logger } from "@trigger.dev/sdk";
+import { z } from "zod";
+
+// Define event with ordering config
+export const testEvent = event({
+  id: "test.greeting",
+  schema: z.object({
+    name: z.string(),
+    message: z.string(),
+  }),
+  ordering: {
+    concurrencyLimit: 2,
+  },
+});
+
+// Slow subscriber with concurrencyLimit: 1 for ordering
+// The concurrencyLimit:1 ensures per-key ordering when used with orderingKey
+export const slowWorker = task({
+  id: "slow-greeting-worker",
+  on: testEvent,
+  queue: { concurrencyLimit: 1 },
+  run: async (payload) => {
+    const start = Date.now();
+    logger.info(`[slow-worker] START "${payload.name}" at ${new Date().toISOString()}`);
+    await new Promise((r) => setTimeout(r, 2000));
+    logger.info(`[slow-worker] END "${payload.name}" after ${Date.now() - start}ms`);
+    return { name: payload.name, duration: Date.now() - start };
+  },
+});