83 limit job data size

src/definitions.ts

@@ -11,4 +11,5 @@ export type Options = {
   };
   redis?: RedisOptions | Redis;
   redisEvents?: RedisOptions | Redis;
+  maxSizeOfJobData?: number /** in bytes */;
 };

src/helpers/roughSizeOfObject.ts

@@ -0,0 +1,36 @@
+import { Options } from '../definitions';
+
+function roughSizeOfObject(object: unknown): number {
+  const objectList = [];
+  const stack = [object];
+  let bytes = 0;
+
+  while (stack.length) {
+    const value = stack.pop();
+
+    if (typeof value === 'boolean') {
+      bytes += 4;
+    } else if (typeof value === 'string') {
+      bytes += value.length * 2;
+    } else if (typeof value === 'number') {
+      bytes += 8;
+    } else if (typeof value === 'object' && objectList.indexOf(value as never) === -1) {
+      objectList.push(value as never);
+
+      for (const i in value) {
+        stack.push(value[i]);
+      }
+    }
+  }
+  return bytes;
+}
+
+export const defaultMaxSizeOfObjectData = 10000;
+
+export function checkJobDataSize(opts: Options, data: unknown): void {
+  const allowableSize = opts?.maxSizeOfJobData || defaultMaxSizeOfObjectData;
+  const size = roughSizeOfObject(data);
+  if (size > allowableSize) {
+    throw new Error(`Job data ${size} exceeds allowable size ${allowableSize}`);
+  }
+}

src/mutation/jobAdd.ts

@@ -3,6 +3,7 @@ import { getJobTC } from '../types/job/Job';
 import { findQueue } from '../helpers';
 import { Options } from '../definitions';
 import { createJobDataITC } from '../types/job/JobInput';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddFC(
   sc: SchemaComposer<any>,
@@ -44,6 +45,7 @@ export function createJobAddFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {

src/mutation/jobAddBulk.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddBulkFC(
   sc: SchemaComposer<any>,
@@ -46,6 +47,14 @@ export function createJobAddBulkFC(
       }).List,
     },
     resolve: async (_, { prefix, queueName, jobs }) => {
+      if (Array.isArray(jobs)) {
+        for (const job of jobs) {
+          checkJobDataSize(opts, job);
+        }
+      } else {
+        throw new Error('jobAddBulk: jobs argument must be an array');
+      }
+
       const queue = await findQueue(prefix, queueName, opts);
       const jobsRes = await queue.addBulk(jobs);
       return {

src/mutation/jobAddCron.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddCronFC(
   sc: SchemaComposer<any>,
@@ -54,6 +55,7 @@ export function createJobAddCronFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {

src/mutation/jobAddEvery.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddEveryFC(
   sc: SchemaComposer<any>,
@@ -53,6 +54,7 @@ export function createJobAddEveryFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {

src/types/queue/Queue.durationAvg.ts

@@ -8,7 +8,7 @@ export function createDurationAvgFC(
   opts: Options
 ): ObjectTypeComposerFieldConfigDefinition<any, any> {
   return {
-    type: 'Int!',
+    type: 'String!',
     args: {
       limit: {
         type: 'Int',
@@ -20,15 +20,15 @@ export function createDurationAvgFC(
       let amount = 0;
       let counter = 0;
       if (jobs.length === 0) {
-        return 0;
+        return '0';
       } else {
         for (const job of jobs) {
           if (job?.finishedOn && job?.processedOn) {
             amount += job.finishedOn - job.processedOn;
             counter++;
           }
         }
-        return (amount / (counter || 1)).toFixed(0);
+        return String((amount / (counter || 1)).toFixed(0));
       }
     },
   };