fix: actually fix ordering again

Author: lindesvard

src/queue.ts

@@ -298,6 +298,25 @@ export type QueueOptions = {
    */
   orderingGracePeriodDecay?: number;
 
+  /**
+   * Maximum number of jobs to collect in a single batch when using `orderingMethod: 'in-memory'`.
+   *
+   * When the grace period expires, the worker collects all remaining jobs in the group up to this limit.
+   * This prevents memory issues and processing delays when many jobs have accumulated.
+   *
+   * @default 10
+   * @example 5 // Smaller batches for faster processing
+   * @example 20 // Larger batches for better throughput
+   * @example 50 // Very large batches for high-volume scenarios
+   *
+   * **When to adjust:**
+   * - High job volume: Increase (20-50) to process more jobs per batch
+   * - Memory constraints: Decrease (5-10) to reduce memory usage
+   * - Processing time: Smaller batches = more responsive, larger batches = better throughput
+   * - Large payloads: Decrease if individual jobs have large data payloads
+   */
+  orderingMaxBatchSize?: number;
+
   /**
    * Number of completed jobs to keep in Redis for inspection and debugging.
    * Older completed jobs are automatically removed to prevent memory growth.
@@ -540,6 +559,7 @@ export class Queue<T = any> {
   private _graceCollectionMs: number; // For 'in-memory' method
   private _orderingMaxWaitMultiplier: number; // Max grace period multiplier
   private _orderingGracePeriodDecay: number; // Grace period decay factor
+  private _orderingMaxBatchSize: number; // Max jobs to collect in a batch
   private keepFailed: number;
   private schedulerLockTtlMs: number;
   public name: string;
@@ -565,6 +585,10 @@ export class Queue<T = any> {
     return this._orderingGracePeriodDecay;
   }
 
+  public get orderingMaxBatchSize(): number {
+    return this._orderingMaxBatchSize;
+  }
+
   // Inline defineCommand bindings removed; using external Lua via evalsha
 
   constructor(opts: QueueOptions) {
@@ -649,6 +673,13 @@ export class Queue<T = any> {
       );
     }
 
+    // Initialize max batch size (default: 10)
+    // Clamp between 1 and 100 to prevent extreme values
+    this._orderingMaxBatchSize = Math.max(
+      1,
+      Math.min(100, opts.orderingMaxBatchSize ?? 10),
+    );
+
     this.r.on('error', (err) => {
       this.logger.error('Redis error (main):', err);
     });

src/worker.ts

@@ -300,6 +300,10 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
   private redisReadyHandler?: () => void;
   private runLoopPromise?: Promise<void>;
 
+  // Track when we started collecting for each group (for in-memory ordering)
+  // This ensures we wait the full grace period from when we FIRST picked up a job from the group
+  private groupCollectionStartTimes = new Map<string, number>();
+
   constructor(opts: WorkerOptions<T>) {
     super();
 
@@ -346,6 +350,10 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
     this.setupRedisEventHandlers();
   }
 
+  get isClosed() {
+    return this.closed;
+  }
+
   /**
    * Add jitter to prevent thundering herd problems in high-concurrency environments
    * @param baseInterval The base interval in milliseconds
@@ -918,6 +926,10 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
    * @param gracefulTimeoutMs Maximum time to wait for current job to finish (default: 30 seconds)
    */
   async close(gracefulTimeoutMs = 30_000): Promise<void> {
+    // Give some time if we just received a job
+    // Otherwise jobsInProgress will be 0 and we will exit immediately
+    await this.delay(100);
+
     this.stopping = true;
 
     if (this.cleanupTimer) {
@@ -1062,12 +1074,18 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
    *
    * How it works:
    * 1. Worker picks up first job and holds it in memory
-   * 2. Waits for the grace period for new jobs to arrive
-   * 3. When a new job arrives, reset the timer and hold ALL jobs for another full grace period
+   * 2. Waits for the grace period to allow concurrent enqueue operations to finish
+   * 3. When new jobs are detected, reset the timer (with decay) and wait again
    * 4. Repeat until no new jobs arrive for the full grace period
    * 5. Process all collected jobs sorted by orderMs
    *
-   * This ensures jobs arriving close together are batched and processed in correct order.
+   * This solves the race condition where jobs are enqueued within 1-2ms:
+   * - Worker reserves job 1 while jobs 2-4 are still being written to Redis
+   * - By always waiting the grace period, we give concurrent writes time to complete
+   * - Then we collect all jobs that finished enqueueing during our wait
+   *
+   * NOTE: The grace window is measured from when WE picked up the job (Date.now()),
+   * NOT from when the job was enqueued. This is critical for handling concurrent enqueues.
    */
   private async collectJobsWithGrace(
     firstJob: ReservedJob<T>,
@@ -1078,60 +1096,94 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
       return [firstJob];
     }
 
+    const groupId = firstJob.groupId;
     const collected: ReservedJob<T>[] = [firstJob];
-    const initialCount = await this.q.getGroupJobCount(firstJob.groupId);
-    let lastSeenCount = initialCount;
-    let lastJobArrivalTime = Date.now(); // Reset every time a new job arrives
-    const startTime = Date.now();
+    const now = Date.now();
+
+    // Check if we're already in a collection window for this group
+    let collectionStartTime = this.groupCollectionStartTimes.get(groupId);
+
+    if (!collectionStartTime) {
+      // First job from this group - start tracking the collection window
+      collectionStartTime = now;
+      this.groupCollectionStartTimes.set(groupId, collectionStartTime);
+
+      this.logger.debug(
+        `Started collection window for group ${groupId} (${initialGraceMs}ms grace period)`,
+      );
+    } else {
+      // We're continuing an existing collection for this group
+      // Check if we've already waited long enough
+      const elapsedSinceStart = now - collectionStartTime;
+      if (elapsedSinceStart >= initialGraceMs) {
+        this.logger.debug(
+          `Collection window for group ${groupId} already expired (${elapsedSinceStart}ms elapsed), processing immediately`,
+        );
+        return [firstJob];
+      }
+
+      this.logger.debug(
+        `Continuing collection for group ${groupId} (${elapsedSinceStart}ms elapsed)`,
+      );
+    }
+
+    let lastSeenCount = 0;
+    let lastJobArrivalTime = now; // Reset every time we detect new jobs
     let currentGraceMs = initialGraceMs; // Decaying grace window
     const minGraceMs = 20; // Minimum wait time
     const decayFactor = this.q.orderingGracePeriodDecay;
 
-    this.logger.debug(
-      `Grace period collection started for group ${firstJob.groupId} (initial count: ${initialCount}, ${initialGraceMs}ms window, decay: ${decayFactor})`,
-    );
+    // ALWAYS wait the grace period when we first pick up a job
+    // This gives concurrent enqueue operations time to finish writing to Redis
+    // The goal is to collect ALL jobs that exist at the END of the grace period
 
     // Keep checking for NEW jobs until grace period expires with no new arrivals
     while (Date.now() - lastJobArrivalTime < currentGraceMs) {
       await this.delay(10); // Check every 10ms
 
-      const currentCount = await this.q.getGroupJobCount(firstJob.groupId);
+      const currentCount = await this.q.getGroupJobCount(groupId);
 
       if (currentCount > lastSeenCount) {
         // New jobs arrived! Reset the grace period timer with decay
         const newJobsCount = currentCount - lastSeenCount;
         lastSeenCount = currentCount;
-        lastJobArrivalTime = Date.now(); // RESET timer
+        lastJobArrivalTime = Date.now(); // RESET timer to when we detected new jobs
 
         // Apply decay: reduce the grace window for the next iteration
         currentGraceMs = Math.max(minGraceMs, currentGraceMs * decayFactor);
 
         this.logger.debug(
-          `${newJobsCount} new job(s) detected in group ${firstJob.groupId}, resetting grace period to ${Math.round(currentGraceMs)}ms (${Math.round((currentGraceMs / initialGraceMs) * 100)}% of original)`,
+          `${newJobsCount} new job(s) detected in group ${groupId}, resetting grace period to ${Math.round(currentGraceMs)}ms (${Math.round((currentGraceMs / initialGraceMs) * 100)}% of original)`,
         );
       }
 
       // Safety: don't wait forever (max multiplier from queue config)
       const maxWaitMs = initialGraceMs * this.q.orderingMaxWaitMultiplier;
-      if (Date.now() - startTime > maxWaitMs) {
+      const totalWaitTime = Date.now() - collectionStartTime;
+      if (totalWaitTime > maxWaitMs) {
         this.logger.warn(
-          `Grace period exceeded ${this.q.orderingMaxWaitMultiplier}x limit (${maxWaitMs}ms) for group ${firstJob.groupId}, proceeding`,
+          `Grace period exceeded ${this.q.orderingMaxWaitMultiplier}x limit (${maxWaitMs}ms) for group ${groupId} (waited ${totalWaitTime}ms), proceeding`,
         );
         break;
       }
     }
 
-    // Only collect jobs we detected DURING the grace period
-    // lastSeenCount is the count at the last detection, which happened within the grace window
-    const jobsArrivedDuringGrace = Math.max(0, lastSeenCount - initialCount);
+    // Collect ALL remaining jobs in the group (up to configured batch limit)
+    // These are the jobs that were there when the grace period ended
+    const finalCount = await this.q.getGroupJobCount(groupId);
+    const maxBatchSize = this.q.orderingMaxBatchSize;
+    const jobsToCollect = Math.min(finalCount, maxBatchSize);
+
+    this.logger.debug(
+      `Grace period complete for group ${groupId}, collecting ${jobsToCollect} of ${finalCount} remaining jobs (max batch: ${maxBatchSize})`,
+    );
 
-    // Only collect jobs that arrived DURING the grace period
-    // Don't collect ALL jobs - that would break the FIFO guarantee
-    for (let i = 0; i < jobsArrivedDuringGrace && i < 20; i++) {
+    // Collect remaining jobs from the group (up to max batch size)
+    for (let i = 0; i < jobsToCollect; i++) {
       try {
         // Reserve next job from this group, passing first job's ID to bypass lock
         const nextJob = await this.reserveNextFromSameGroup(
-          firstJob.groupId,
+          groupId,
           firstJob.id,
         );
         if (!nextJob) break;
@@ -1155,7 +1207,7 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
 
     if (collected.length > 1) {
       this.logger.info(
-        `Collected ${collected.length} jobs from group ${firstJob.groupId} during ${Date.now() - startTime}ms grace period`,
+        `Collected ${collected.length} jobs from group ${groupId} during ${Date.now() - now}ms grace period`,
         {
           unsorted: oldCollected.map((job) => job.id),
           sorted: collected.map((job, index) => ({
@@ -1169,6 +1221,10 @@ class _Worker<T = any> extends TypedEventEmitter<WorkerEvents<T>> {
       );
     }
 
+    // Clear the tracking for this group after we've collected the batch
+    // This allows the next batch to start fresh
+    this.groupCollectionStartTimes.delete(groupId);
+
     return collected;
   }
 

test/queue.graceful-shutdown.test.ts

@@ -254,7 +254,7 @@ describe('Graceful Shutdown Tests', () => {
     const elapsed = Date.now() - startTime;
 
     expect(elapsed).toBeGreaterThan(190);
-    expect(elapsed).toBeLessThan(500); // Increased tolerance for proper run loop shutdown
+    expect(elapsed).toBeLessThan(600);
     expect(sawGracefulTimeout).toBe(true);
 
     shouldStop = true; // Allow the handler to finish
@@ -481,4 +481,34 @@ describe('Graceful Shutdown Tests', () => {
 
     await redis.quit();
   }, 10000); // 10 second timeout for the test
+
+  it('should shutdown gracefully when we have orderingWindowMs + orderingMethod = "in-memory"', async () => {
+    const redis = new Redis(REDIS_URL);
+    const queue = new Queue({
+      redis,
+      logger: true,
+      namespace: `${namespace}:in-memory`,
+      orderingMethod: 'in-memory',
+      orderingWindowMs: 1000,
+    });
+    let isCompleted = false;
+    const worker = new Worker({
+      queue: queue,
+      logger: true,
+      handler: async (job) => {
+        await new Promise((resolve) => setTimeout(resolve, 1000));
+        isCompleted = true;
+      },
+    });
+    await queue.add({ groupId: 'test-group', data: { id: 1 } });
+    worker.on('completed', (job) => {
+      console.log('Completed', job.id);
+    });
+    worker.run();
+    await worker.close(2000);
+    expect(isCompleted).toBe(true);
+    expect(worker.isProcessing()).toBe(false);
+    expect(worker.getCurrentJob()).toBe(null);
+    expect(worker.isClosed).toBe(true);
+  });
 });

test/queue.ordering-grace.test.ts

@@ -251,6 +251,64 @@ describe('Ordering Grace Period (orderingGraceMs)', () => {
 
     await worker.close();
   });
+
+  it('should collect jobs added within 1-2ms (race condition test)', async () => {
+    const q = new Queue({
+      redis,
+      namespace: `${namespace}:race-condition`,
+      jobTimeoutMs: 5000,
+      orderingMethod: 'in-memory',
+      orderingWindowMs: 100, // Grace period to allow concurrent adds to complete
+      orderingGracePeriodDecay: 0.9,
+      orderingMaxWaitMultiplier: 8,
+      logger: true, // Disable verbose logging
+    });
+
+    const collectCalls: number[] = []; // Track how many jobs collected in each call
+    const processed: number[] = [];
+
+    const worker = new Worker<{ seq: number }>({
+      queue: q,
+      concurrency: 6,
+      logger: true,
+      handler: async (job) => {
+        processed.push(job.data.seq);
+        // Don't do atomic completion - process jobs slower to avoid
+        // the next job being grabbed before we can see the collection
+        await wait(50); // Longer than grace period to ensure we see batching
+      },
+    });
+    worker.run();
+
+    const baseTime = Date.now();
+
+    // Add all 4 jobs concurrently (within ~1ms)
+    // This simulates the real-world scenario where multiple API calls
+    // trigger job creation almost simultaneously
+    console.log(
+      'Adding jobs within 1ms window (simulating concurrent API calls)...',
+    );
+    await q.add({
+      groupId: 'device-1',
+      data: { seq: 3 },
+      orderMs: baseTime + 2,
+    });
+    await wait(50);
+    await Promise.all([
+      q.add({ groupId: 'device-1', data: { seq: 2 }, orderMs: baseTime + 1 }),
+      q.add({ groupId: 'device-1', data: { seq: 1 }, orderMs: baseTime }),
+      q.add({ groupId: 'device-1', data: { seq: 4 }, orderMs: baseTime + 3 }),
+    ]);
+    console.log('All 4 jobs added');
+
+    await q.waitForEmpty();
+    await wait(100);
+
+    // All jobs should be processed in order
+    expect(processed).toEqual([1, 2, 3, 4]);
+
+    await worker.close();
+  });
 });
 
 async function wait(ms: number) {

test/queue.stress.test.ts

@@ -217,7 +217,7 @@ describe('Stress and Performance Degradation Tests', () => {
     // In production, worker churn would be much less aggressive.
     expect(duplicateRate).toBeLessThan(0.1); // Less than 10% duplicates
 
-    await redis.quit();
+    // await redis.quit();
   }, 30000);
 
   it('should handle burst traffic patterns', async () => {