improve polling behavior, at least 0.5 if no long polling supported

Author: muniter

src/core/queue.ts

@@ -33,6 +33,13 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
   protected plugins: QueuePlugin[];
   protected pluginDisposers: Array<() => Promise<void>> = [];
   public readonly name?: string;
+  
+  /**
+   * Indicates whether this queue driver supports long polling.
+   * Drivers that support long polling (like SQS) can efficiently wait for messages.
+   * Drivers that don't support long polling will have a minimum 0.5s sleep between polls.
+   */
+  protected supportsLongPolling = false;
 
   /**
    * Creates a new Queue instance.
@@ -195,8 +202,14 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
         const message = await this.reserve(timeout);
         if (!message) {
           if (!repeat) break;
-          if (timeout > 0) {
-            await this.sleep(timeout * 1000);
+          
+          // Apply minimum sleep time for drivers that don't support long polling
+          const sleepMs = this.supportsLongPolling 
+            ? timeout * 1000 
+            : Math.max(500, timeout * 1000);
+          
+          if (sleepMs > 0) {
+            await this.sleep(sleepMs);
           }
           continue;
         }
@@ -326,7 +339,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
     return true; // Job is considered handled (failed)
   }
 
-  private async sleep(ms: number): Promise<void> {
+  protected async sleep(ms: number): Promise<void> {
     return new Promise(resolve => setTimeout(resolve, ms));
   }
 

src/drivers/sqs.ts

@@ -28,6 +28,8 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
     options: QueueOptions = {}
   ) {
     super(options);
+    // SQS supports long polling via WaitTimeSeconds
+    this.supportsLongPolling = true;
   }
 
   protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {

tests/core/queue.test.ts

@@ -38,6 +38,13 @@ class TestQueue extends Queue<TestJobs, DbJobRequest<any>> {
   }
 }
 
+class TestQueueWithLongPolling extends TestQueue {
+  constructor(options = {}) {
+    super(options);
+    this.supportsLongPolling = true;
+  }
+}
+
 describe('Queue', () => {
   let queue: TestQueue;
 
@@ -221,4 +228,110 @@ describe('Queue', () => {
       );
     });
   });
+
+  describe('polling behavior', () => {
+    // Plugin to track polling times
+    class PollTrackingPlugin {
+      public pollTimes: number[] = [];
+      public pollCount = 0;
+      
+      async beforePoll(): Promise<'continue' | 'stop'> {
+        this.pollTimes.push(Date.now());
+        this.pollCount++;
+        
+        // Stop after a few polls to prevent infinite loop
+        if (this.pollCount >= 3) {
+          return 'stop';
+        }
+        return 'continue';
+      }
+      
+      getTimeBetweenPolls(index1: number, index2: number): number {
+        if (index1 >= this.pollTimes.length || index2 >= this.pollTimes.length) {
+          return -1;
+        }
+        return this.pollTimes[index2]! - this.pollTimes[index1]!;
+      }
+    }
+
+    it('should sleep for at least 500ms when timeout is 0 (non-long-polling)', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueue({ plugins: [plugin] });
+      queue.onJob('test-job', async () => {});
+      
+      // Run with repeat=true and timeout=0
+      await queue.run(true, 0);
+      
+      // Check time between polls
+      const timeBetweenPolls = plugin.getTimeBetweenPolls(0, 1);
+      expect(timeBetweenPolls).toBeGreaterThanOrEqual(500);
+      expect(timeBetweenPolls).toBeLessThan(600); // Allow some margin
+    });
+
+    it('should sleep for at least 500ms when timeout < 0.5s (non-long-polling)', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueue({ plugins: [plugin] });
+      queue.onJob('test-job', async () => {});
+      
+      // Run with timeout=0.2 (200ms)
+      await queue.run(true, 0.2);
+      
+      // Should still sleep for 500ms, not 200ms
+      const timeBetweenPolls = plugin.getTimeBetweenPolls(0, 1);
+      expect(timeBetweenPolls).toBeGreaterThanOrEqual(500);
+      expect(timeBetweenPolls).toBeLessThan(600);
+    });
+
+    it('should use timeout value when > 0.5s (non-long-polling)', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueue({ plugins: [plugin] });
+      queue.onJob('test-job', async () => {});
+      
+      // Run with timeout=2 (2 seconds)
+      await queue.run(true, 2);
+      
+      // Should sleep for 2000ms
+      const timeBetweenPolls = plugin.getTimeBetweenPolls(0, 1);
+      expect(timeBetweenPolls).toBeGreaterThanOrEqual(2000);
+      expect(timeBetweenPolls).toBeLessThan(2100);
+    });
+
+    it('should not sleep when timeout is 0 (long-polling)', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueueWithLongPolling({ plugins: [plugin] });
+      queue.onJob('test-job', async () => {});
+      
+      // Run with timeout=0
+      await queue.run(true, 0);
+      
+      // With long polling and timeout=0, polls should be immediate
+      const timeBetweenPolls = plugin.getTimeBetweenPolls(0, 1);
+      expect(timeBetweenPolls).toBeLessThan(50); // Should be nearly instant
+    });
+
+    it('should use exact timeout value (long-polling)', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueueWithLongPolling({ plugins: [plugin] });
+      queue.onJob('test-job', async () => {});
+      
+      // Run with timeout=0.2 (200ms)
+      await queue.run(true, 0.2);
+      
+      // Should sleep for exactly 200ms
+      const timeBetweenPolls = plugin.getTimeBetweenPolls(0, 1);
+      expect(timeBetweenPolls).toBeGreaterThanOrEqual(200);
+      expect(timeBetweenPolls).toBeLessThan(250);
+    });
+
+    it('should not sleep in single run mode', async () => {
+      const plugin = new PollTrackingPlugin();
+      const queue = new TestQueue({ plugins: [plugin] });
+      
+      // Run with repeat=false
+      await queue.run(false, 0);
+      
+      // Should only poll once
+      expect(plugin.pollCount).toBe(1);
+    });
+  });
 });