feat: limit how many tasks are executing at any time (#1103)

Allows user defined concurrency limit when evaluating check alert tasks. Includes some further refinements of the task execution types to make actual task execution more type safe.

Author: dhable

.changeset/six-squids-stare.md

@@ -0,0 +1,5 @@
+---
+"@hyperdx/api": patch
+---
+
+feat: limit how many tasks are executing at any time

packages/api/package.json

@@ -8,6 +8,7 @@
   },
   "dependencies": {
     "@clickhouse/client": "^0.2.10",
+    "@esm2cjs/p-queue": "^7.3.0",
     "@hyperdx/common-utils": "^0.3.2",
     "@hyperdx/lucene": "^3.1.1",
     "@hyperdx/node-opentelemetry": "^0.8.2",

packages/api/src/tasks/__tests__/types.test.ts

@@ -14,7 +14,8 @@ describe('asTaskArgs', () => {
       provider: 'default',
     });
     expect(result.taskName).toBe('command');
-    expect(result.provider).toBe('default');
+    // For non-check-alerts tasks, we need to use type assertion to access provider
+    expect((result as any).provider).toBe('default');
   });
 
   it('should return valid TaskArgs when provider is undefined', () => {
@@ -29,18 +30,8 @@ describe('asTaskArgs', () => {
       provider: undefined,
     });
     expect(result.taskName).toBe('command');
-    expect(result.provider).toBeUndefined();
-  });
-
-  it('should throw error when provider is empty string', () => {
-    const invalidArgs = {
-      _: ['command'],
-      provider: '',
-    };
-
-    expect(() => asTaskArgs(invalidArgs)).toThrow(
-      'Provider must contain valid characters',
-    );
+    // For non-check-alerts tasks, we need to use type assertion to access provider
+    expect((result as any).provider).toBeUndefined();
   });
 
   it('should throw error for null input', () => {
@@ -95,21 +86,6 @@ describe('asTaskArgs', () => {
     );
   });
 
-  it('should throw error when provider is not a string', () => {
-    const invalidProviders = [123, null, { name: 'default' }, ['default']];
-
-    invalidProviders.forEach(provider => {
-      const invalidArgs = {
-        _: ['command'],
-        provider,
-      };
-
-      expect(() => asTaskArgs(invalidArgs)).toThrow(
-        'Provider must be a string if provided',
-      );
-    });
-  });
-
   it('should handle empty array for _ property', () => {
     const validArgs = {
       _: [],
@@ -169,6 +145,11 @@ describe('asTaskArgs', () => {
       taskName: 'check-alerts',
       provider: 'default',
     });
+    expect(result.taskName).toBe('check-alerts');
+    // For check-alerts tasks, provider property is directly accessible
+    if (result.taskName === 'check-alerts') {
+      expect(result.provider).toBe('default');
+    }
   });
 
   it('should accept check-alerts task without provider', () => {
@@ -182,16 +163,257 @@ describe('asTaskArgs', () => {
       taskName: 'check-alerts',
       provider: undefined,
     });
+    expect(result.taskName).toBe('check-alerts');
+    // For check-alerts tasks, provider property is directly accessible
+    if (result.taskName === 'check-alerts') {
+      expect(result.provider).toBeUndefined();
+    }
   });
 
-  it('should throw error when provider is whitespace-only', () => {
-    const invalidArgs = {
-      _: ['command'],
-      provider: '   ',
+  it('should accept ping-pong task without provider', () => {
+    const validArgs = {
+      _: ['ping-pong'],
     };
 
-    expect(() => asTaskArgs(invalidArgs)).toThrow(
-      'Provider must contain valid characters',
-    );
+    const result = asTaskArgs(validArgs);
+
+    expect(result).toEqual({
+      taskName: 'ping-pong',
+    });
+    expect(result.taskName).toBe('ping-pong');
+    // Ping-pong tasks should not have a provider property
+    expect('provider' in result).toBe(false);
+  });
+
+  describe('concurrency parameter validation', () => {
+    it('should accept check-alerts task with valid concurrency', () => {
+      const validArgs = {
+        _: ['check-alerts'],
+        concurrency: 4,
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'check-alerts',
+        provider: undefined,
+        concurrency: 4,
+      });
+      expect(result.taskName).toBe('check-alerts');
+      if (result.taskName === 'check-alerts') {
+        expect(result.concurrency).toBe(4);
+      }
+    });
+
+    it('should accept check-alerts task with concurrency value of 1', () => {
+      const validArgs = {
+        _: ['check-alerts'],
+        concurrency: 1,
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'check-alerts',
+        provider: undefined,
+        concurrency: 1,
+      });
+      expect(result.taskName).toBe('check-alerts');
+      if (result.taskName === 'check-alerts') {
+        expect(result.concurrency).toBe(1);
+      }
+    });
+
+    it('should accept check-alerts task with large concurrency values', () => {
+      const validArgs = {
+        _: ['check-alerts'],
+        concurrency: 100,
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'check-alerts',
+        provider: undefined,
+        concurrency: 100,
+      });
+      expect(result.taskName).toBe('check-alerts');
+      if (result.taskName === 'check-alerts') {
+        expect(result.concurrency).toBe(100);
+      }
+    });
+
+    it('should accept check-alerts task without concurrency parameter', () => {
+      const validArgs = {
+        _: ['check-alerts'],
+        provider: 'default',
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'check-alerts',
+        provider: 'default',
+        concurrency: undefined,
+      });
+      expect(result.taskName).toBe('check-alerts');
+      if (result.taskName === 'check-alerts') {
+        expect(result.concurrency).toBeUndefined();
+      }
+    });
+
+    it('should throw error when concurrency is not a number', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: 'invalid',
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be a number if provided',
+      );
+    });
+
+    it('should throw error when concurrency is boolean', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: true,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be a number if provided',
+      );
+    });
+
+    it('should throw error when concurrency is null', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: null,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be a number if provided',
+      );
+    });
+
+    it('should throw error when concurrency is an object', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: { value: 4 },
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be a number if provided',
+      );
+    });
+
+    it('should throw error when concurrency is an array', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: [4],
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be a number if provided',
+      );
+    });
+
+    it('should throw error when concurrency is zero', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: 0,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency cannot be less than 1',
+      );
+    });
+
+    it('should throw error when concurrency is negative', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: -1,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency cannot be less than 1',
+      );
+    });
+
+    it('should throw error when concurrency is a negative decimal', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: -0.5,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be an integer if provided',
+      );
+    });
+
+    it('should throw error when concurrency is a positive decimal', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: 2.5,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be an integer if provided',
+      );
+    });
+
+    it('should throw error when concurrency is a small decimal', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: 1.1,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be an integer if provided',
+      );
+    });
+
+    it('should throw error when concurrency is a large decimal', () => {
+      const invalidArgs = {
+        _: ['check-alerts'],
+        concurrency: 100.999,
+      };
+
+      expect(() => asTaskArgs(invalidArgs)).toThrow(
+        'Concurrency must be an integer if provided',
+      );
+    });
+
+    it('should ignore concurrency parameter for non-check-alerts tasks', () => {
+      const validArgs = {
+        _: ['ping-pong'],
+        concurrency: 4,
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'ping-pong',
+      });
+      expect(result.taskName).toBe('ping-pong');
+      // Ping-pong tasks should not have a concurrency property
+      expect('concurrency' in result).toBe(false);
+    });
+
+    it('should ignore concurrency parameter for unknown task types', () => {
+      const validArgs = {
+        _: ['unknown-task'],
+        concurrency: 4,
+      };
+
+      const result = asTaskArgs(validArgs);
+
+      expect(result).toEqual({
+        taskName: 'unknown-task',
+        provider: undefined,
+      });
+      expect(result.taskName).toBe('unknown-task');
+      // Unknown task types should not process concurrency parameter
+      expect('concurrency' in result).toBe(false);
+    });
   });
 });