feat(events): ordering key with per-key serialization + global concurrency limit

Run engine changes:
- Add globalConcurrencyLimit (gcl) and globalCurrentConcurrency (gcc) Redis keys
- Modify dequeueMessagesFromQueue Lua: check global limit when gcl exists
- Modify releaseConcurrency Lua: SREM from gcc set
- Modify enqueueMessage/enqueueMessageWithTtl: SREM from gcc on re-enqueue
- No impact on existing queues (gcl check only runs when key exists)

Event system changes:
- PublishEventService overrides queue to `evt-order:{eventSlug}` when orderingKey present
- Deploy creates ordering queue with concurrencyLimit:1 (per-key) + global limit
- SDK event() accepts `ordering: { concurrencyLimit: N }` config
- EventManifest/EventMetadata include ordering field

Behavior: orderingKey guarantees strict per-key ordering (1 at a time per key)
while concurrencyLimit controls total parallel runs across all keys.

Run-engine tests: 236 pass, 2 fail (pre-existing flaky, not caused by this change)
Event integration tests: 24/24 pass

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

Author: giovaborgogno

apps/webapp/app/v3/runQueue.server.ts

@@ -32,6 +32,15 @@ export async function updateQueueConcurrencyLimits(
   ]);
 }
 
+/** Updates the global concurrency limit for a queue (across all concurrency keys) */
+export async function updateGlobalQueueConcurrencyLimits(
+  environment: AuthenticatedEnvironment,
+  queueName: string,
+  concurrency: number
+) {
+  await engine.runQueue.updateGlobalQueueConcurrencyLimits(environment, queueName, concurrency);
+}
+
 /** Removes MARQS and the RunQueue limits for a queue */
 export async function removeQueueConcurrencyLimits(
   environment: AuthenticatedEnvironment,

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

@@ -18,6 +18,7 @@ import {
   removeQueueConcurrencyLimits,
   updateEnvConcurrencyLimits,
   updateQueueConcurrencyLimits,
+  updateGlobalQueueConcurrencyLimits,
 } from "../runQueue.server";
 import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion";
 import { clampMaxDuration } from "../utils/maxDuration";
@@ -362,6 +363,26 @@ async function syncWorkerEvents(
       });
 
       eventDefinitions.set(event.id, eventDef.id);
+
+      // Create ordering queue for events that have ordering config
+      if (event.ordering) {
+        const orderingQueueName = `evt-order:${event.id}`;
+        const globalLimit = event.ordering.concurrencyLimit ?? environment.maximumConcurrencyLimit;
+
+        await upsertWorkerQueueRecord(
+          orderingQueueName,
+          1, // per-key limit: always 1 for strict ordering
+          orderingQueueName,
+          "SHARED",
+          worker,
+          prisma
+        );
+
+        // Set per-key limit = 1 in Redis
+        await updateQueueConcurrencyLimits(environment, orderingQueueName, 1);
+        // Set global limit in Redis (total concurrent across all keys)
+        await updateGlobalQueueConcurrencyLimits(environment, orderingQueueName, globalLimit);
+      }
     }
   }
 

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

@@ -303,6 +303,11 @@ 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

@@ -341,6 +341,21 @@ export class RunQueue {
     return this.redis.del(this.keys.queueConcurrencyLimitKey(env, queue));
   }
 
+  public async updateGlobalQueueConcurrencyLimits(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string,
+    concurrency: number
+  ) {
+    return this.redis.set(this.keys.queueGlobalConcurrencyLimitKey(env, queue), concurrency);
+  }
+
+  public async removeGlobalQueueConcurrencyLimits(
+    env: MinimalAuthenticatedEnvironment,
+    queue: string
+  ) {
+    return this.redis.del(this.keys.queueGlobalConcurrencyLimitKey(env, queue));
+  }
+
   public async getQueueConcurrencyLimit(env: MinimalAuthenticatedEnvironment, queue: string) {
     const result = await this.redis.get(this.keys.queueConcurrencyLimitKey(env, queue));
 
@@ -906,6 +921,7 @@ export class RunQueue {
           this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
           this.keys.queueCurrentDequeuedKeyFromQueue(message.queue),
           this.keys.envCurrentDequeuedKeyFromQueue(message.queue),
+          this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(message.queue),
           messageId
         );
       },
@@ -1655,6 +1671,7 @@ export class RunQueue {
     const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKeyFromQueue(message.queue);
     const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKeyFromQueue(message.queue);
     const envQueueKey = this.keys.envQueueKeyFromQueue(message.queue);
+    const globalCurrentConcurrencyKey = this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(message.queue);
     const masterQueueKey = this.keys.masterQueueKeyForEnvironment(
       message.environmentId,
       this.shardCount
@@ -1694,6 +1711,7 @@ export class RunQueue {
         envCurrentDequeuedKey,
         envQueueKey,
         ttlInfo.ttlQueueKey,
+        globalCurrentConcurrencyKey,
         queueName,
         messageId,
         messageData,
@@ -1711,6 +1729,7 @@ export class RunQueue {
         queueCurrentDequeuedKey,
         envCurrentDequeuedKey,
         envQueueKey,
+        globalCurrentConcurrencyKey,
         queueName,
         messageId,
         messageData,
@@ -1745,6 +1764,10 @@ export class RunQueue {
       const messageKeyPrefix = this.keys.messageKeyPrefixFromQueue(messageQueue);
       const envQueueKey = this.keys.envQueueKeyFromQueue(messageQueue);
       const masterQueueKey = this.keys.masterQueueKeyForShard(shard);
+      const globalConcurrencyLimitKey =
+        this.keys.queueGlobalConcurrencyLimitKeyFromQueue(messageQueue);
+      const globalCurrentConcurrencyKey =
+        this.keys.queueGlobalCurrentConcurrencyKeyFromQueue(messageQueue);
 
       // Get TTL queue key if TTL system is enabled
       const ttlShardCount = this.options.ttlSystem?.shardCount ?? this.shardCount;
@@ -1767,6 +1790,8 @@ export class RunQueue {
         envQueueKey,
         masterQueueKey,
         ttlQueueKey,
+        globalConcurrencyLimitKey,
+        globalCurrentConcurrencyKey,
         shard,
         maxCount,
       });
@@ -1783,6 +1808,8 @@ export class RunQueue {
         envQueueKey,
         masterQueueKey,
         ttlQueueKey,
+        globalConcurrencyLimitKey,
+        globalCurrentConcurrencyKey,
         //args
         messageQueue,
         String(Date.now()),
@@ -2502,7 +2529,7 @@ end
     });
 
     this.redis.defineCommand("enqueueMessage", {
-      numberOfKeys: 8,
+      numberOfKeys: 9,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -2512,6 +2539,7 @@ local envCurrentConcurrencyKey = KEYS[5]
 local queueCurrentDequeuedKey = KEYS[6]
 local envCurrentDequeuedKey = KEYS[7]
 local envQueueKey = KEYS[8]
+local globalCurrentConcurrencyKey = KEYS[9]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -2541,12 +2569,13 @@ redis.call('SREM', queueCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
       `,
     });
 
     // Enqueue with TTL tracking - atomically adds to both normal queue and TTL sorted set
     this.redis.defineCommand("enqueueMessageWithTtl", {
-      numberOfKeys: 9,
+      numberOfKeys: 10,
       lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -2557,6 +2586,7 @@ local queueCurrentDequeuedKey = KEYS[6]
 local envCurrentDequeuedKey = KEYS[7]
 local envQueueKey = KEYS[8]
 local ttlQueueKey = KEYS[9]
+local globalCurrentConcurrencyKey = KEYS[10]
 
 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -2591,6 +2621,7 @@ redis.call('SREM', queueCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
       `,
     });
 
@@ -2692,7 +2723,7 @@ return results
     });
 
     this.redis.defineCommand("dequeueMessagesFromQueue", {
-      numberOfKeys: 10,
+      numberOfKeys: 12,
       lua: `
 local queueKey = KEYS[1]
 local queueConcurrencyLimitKey = KEYS[2]
@@ -2704,6 +2735,8 @@ local messageKeyPrefix = KEYS[7]
 local envQueueKey = KEYS[8]
 local masterQueueKey = KEYS[9]
 local ttlQueueKey = KEYS[10]  -- Optional: TTL sorted set key (empty string if not used)
+local globalConcurrencyLimitKey = KEYS[11]  -- Global queue concurrency limit (without :ck:)
+local globalCurrentConcurrencyKey = KEYS[12]  -- Global queue concurrency tracking (without :ck:)
 
 local queueName = ARGV[1]
 local currentTime = tonumber(ARGV[2])
@@ -2722,7 +2755,7 @@ if envCurrentConcurrency >= envConcurrencyLimitWithBurstFactor then
     return nil
 end
 
--- Check current queue concurrency against the limit
+-- Check current queue concurrency against the limit (per-key when concurrencyKey is used)
 local queueCurrentConcurrency = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
 local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envConcurrencyLimit)
 local totalQueueConcurrencyLimit = queueConcurrencyLimit
@@ -2732,10 +2765,23 @@ if queueCurrentConcurrency >= totalQueueConcurrencyLimit then
     return nil
 end
 
+-- Check global queue concurrency limit (across all concurrency keys)
+-- Only applies when globalConcurrencyLimitKey is set (e.g. for event ordering queues)
+local globalAvailableCapacity = 1000000
+local globalConcurrencyLimitRaw = redis.call('GET', globalConcurrencyLimitKey)
+if globalConcurrencyLimitRaw then
+    local globalConcurrencyLimit = tonumber(globalConcurrencyLimitRaw)
+    local globalCurrentConcurrency = tonumber(redis.call('SCARD', globalCurrentConcurrencyKey) or '0')
+    if globalCurrentConcurrency >= globalConcurrencyLimit then
+        return nil
+    end
+    globalAvailableCapacity = globalConcurrencyLimit - globalCurrentConcurrency
+end
+
 -- Calculate how many messages we can actually dequeue based on concurrency limits
 local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency
 local queueAvailableCapacity = totalQueueConcurrencyLimit - queueCurrentConcurrency
-local actualMaxCount = math.min(maxCount, envAvailableCapacity, queueAvailableCapacity)
+local actualMaxCount = math.min(maxCount, envAvailableCapacity, queueAvailableCapacity, globalAvailableCapacity)
 
 if actualMaxCount <= 0 then
     return nil
@@ -2779,6 +2825,11 @@ for i = 1, #messages, 2 do
             redis.call('SADD', queueCurrentConcurrencyKey, messageId)
             redis.call('SADD', envCurrentConcurrencyKey, messageId)
 
+            -- Track global queue concurrency (only if global limit is configured)
+            if globalConcurrencyLimitRaw then
+                redis.call('SADD', globalCurrentConcurrencyKey, messageId)
+            end
+
             -- Remove from TTL set if provided (run is being executed, not expired)
             if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then
                 local ttlMember = queueName .. '|' .. messageId .. '|' .. (messageData.orgId or '')
@@ -2999,13 +3050,14 @@ redis.call('SREM', envCurrentDequeuedKey, messageId)
     });
 
     this.redis.defineCommand("releaseConcurrency", {
-      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]
@@ -3015,6 +3067,7 @@ redis.call('SREM', queueCurrentConcurrencyKey, messageId)
 redis.call('SREM', envCurrentConcurrencyKey, messageId)
 redis.call('SREM', queueCurrentDequeuedKey, messageId)
 redis.call('SREM', envCurrentDequeuedKey, messageId)
+redis.call('SREM', globalCurrentConcurrencyKey, messageId)
 `,
     });
 
@@ -3139,6 +3192,7 @@ declare module "@internal/redis" {
       queueCurrentDequeuedKey: string,
       envCurrentDequeuedKey: string,
       envQueueKey: string,
+      globalCurrentConcurrencyKey: string,
       //args
       queueName: string,
       messageId: string,
@@ -3158,6 +3212,7 @@ declare module "@internal/redis" {
       envCurrentDequeuedKey: string,
       envQueueKey: string,
       ttlQueueKey: string,
+      globalCurrentConcurrencyKey: string,
       //args
       queueName: string,
       messageId: string,
@@ -3194,6 +3249,8 @@ declare module "@internal/redis" {
       envQueueKey: string,
       masterQueueKey: string,
       ttlQueueKey: string,
+      globalConcurrencyLimitKey: string,
+      globalCurrentConcurrencyKey: string,
       //args
       childQueueName: string,
       currentTime: string,
@@ -3288,6 +3345,7 @@ declare module "@internal/redis" {
       envCurrentConcurrencyKey: string,
       queueCurrentDequeuedKey: string,
       envCurrentDequeuedKey: string,
+      globalCurrentConcurrencyKey: string,
       // args
       messageId: string,
       callback?: Callback<void>

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

@@ -17,6 +17,8 @@ const constants = {
   DEAD_LETTER_QUEUE_PART: "deadLetter",
   MASTER_QUEUE_PART: "masterQueue",
   WORKER_QUEUE_PART: "workerQueue",
+  GLOBAL_CONCURRENCY_LIMIT_PART: "globalConcurrency",
+  GLOBAL_CURRENT_CONCURRENCY_PART: "globalCurrentConcurrency",
 } as const;
 
 export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
@@ -138,6 +140,20 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
     return `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`;
   }
 
+  queueGlobalConcurrencyLimitKeyFromQueue(queue: string) {
+    const globalQueueName = queue.replace(/:ck:.+$/, "");
+    return `${globalQueueName}:${constants.GLOBAL_CONCURRENCY_LIMIT_PART}`;
+  }
+
+  queueGlobalCurrentConcurrencyKeyFromQueue(queue: string) {
+    const globalQueueName = queue.replace(/:ck:.+$/, "");
+    return `${globalQueueName}:${constants.GLOBAL_CURRENT_CONCURRENCY_PART}`;
+  }
+
+  queueGlobalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string) {
+    return [this.queueKey(env, queue), constants.GLOBAL_CONCURRENCY_LIMIT_PART].join(":");
+  }
+
   queueCurrentConcurrencyKeyFromQueue(queue: string) {
     return `${queue}:${constants.CURRENT_CONCURRENCY_PART}`;
   }

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

@@ -75,6 +75,9 @@ export interface RunQueueKeyProducer {
   envQueueKeyFromQueue(queue: string): string;
   queueConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
   queueConcurrencyLimitKeyFromQueue(queue: string): string;
+  queueGlobalConcurrencyLimitKeyFromQueue(queue: string): string;
+  queueGlobalCurrentConcurrencyKeyFromQueue(queue: string): string;
+  queueGlobalConcurrencyLimitKey(env: RunQueueKeyProducerEnvironment, queue: string): string;
   queueCurrentConcurrencyKeyFromQueue(queue: string): string;
   queueCurrentConcurrencyKey(
     env: RunQueueKeyProducerEnvironment,

packages/core/src/v3/resource-catalog/catalog.ts

@@ -9,6 +9,8 @@ export interface EventMetadata {
   rawSchema?: unknown;
   /** Rate limit configuration */
   rateLimit?: { limit: number; window: string };
+  /** Ordering configuration — enables per-key serialization with global concurrency limit */
+  ordering?: { concurrencyLimit?: number };
 }
 
 export interface ResourceCatalog {

packages/core/src/v3/resource-catalog/standardResourceCatalog.ts

@@ -191,6 +191,7 @@ export class StandardResourceCatalog implements ResourceCatalog {
       version: event.version,
       description: event.description,
       rateLimit: event.rateLimit,
+      ordering: event.ordering,
     }));
   }
 

packages/core/src/v3/schemas/schemas.ts

@@ -183,6 +183,13 @@ export const EventRateLimitManifest = z.object({
 
 export type EventRateLimitManifest = z.infer<typeof EventRateLimitManifest>;
 
+export const EventOrderingManifest = z.object({
+  /** Maximum number of ordering keys processed in parallel */
+  concurrencyLimit: z.number().int().positive().optional(),
+});
+
+export type EventOrderingManifest = z.infer<typeof EventOrderingManifest>;
+
 export const EventManifest = z.object({
   /** Unique event identifier (e.g. "order.created") */
   id: z.string(),
@@ -194,6 +201,8 @@ export const EventManifest = z.object({
   schema: z.unknown().optional(),
   /** Rate limit configuration */
   rateLimit: EventRateLimitManifest.optional(),
+  /** Ordering configuration — creates a dedicated queue with per-key serialization */
+  ordering: EventOrderingManifest.optional(),
 });
 
 export type EventManifest = z.infer<typeof EventManifest>;