feat: enhance task logging and job options in QueueManager - Add logging of parameters in MonkeyCapture decorator for better traceability - Update TaskGroup logging level from debug to info for improved clarity - Introduce new job options in QueueManager for better task management, including removeOnFail and removeDependencyOnFailure - Refactor Worker class to utilize updated job options and handle timeouts more effectively

Author: theboringhumane

packages/core/src/decorators/monkeyLog.ts

@@ -17,6 +17,18 @@ export function MonkeyCapture(fn: Function) {
 
     const taskHistoryKey = `${WORKER_KEY}:${workerId}:task:${job.id}:logs`;
 
+    const parameters = args[3];
+
+    redis.lpush(
+      taskHistoryKey,
+      JSON.stringify({
+        timestamp: new Date().toISOString(),
+        level: "info",
+        message: "parameters",
+        functionArgs: parameters,
+      })
+    );
+
     // grep all the internal functions inside the wrapped function
     const internalFunctions = fn.toString().match(/function\s+(\w+)\s*\(/g);
 

packages/core/src/groups/taskGroup.ts

@@ -696,7 +696,7 @@ export class TaskGroup {
 
     // In client mode, there's no worker and no processing needed
     if (!this.worker) {
-      logger.debug("🔄 TaskGroup: No worker available (client mode), skipping processing", {
+      logger.info("🔄 TaskGroup: No worker available (client mode), skipping processing", {
         file: "taskGroup.ts",
         function: "startProcessing",
         groupName: this.config.name,

packages/core/src/queue/queueManager.ts

@@ -1,4 +1,4 @@
-import { Queue, QueueOptions, QueueEvents } from "bullmq";
+import { Queue, QueueOptions, QueueEvents, JobsOptions } from "bullmq";
 import { Task, TaskOptions, WorkerConfig } from "../types/interfaces";
 import {
   TaskState,
@@ -59,8 +59,6 @@ export class QueueManager {
     this.metrics = new QueueMetrics(this.redis);
     this.shouldCreateWorkers = createWorkers;
 
-    console.log("🔍 QueueManager constructor:", { defaultQueueName, createWorkers, shouldCreateWorkers: this.shouldCreateWorkers });
-
     const finalQueueOptions: QueueOptions = {
       connection: this.redis.options,
       ...queueOptions,
@@ -411,22 +409,32 @@ export class QueueManager {
       updatedAt: new Date(),
     };
 
-    const jobOptions = {
-      id: task.id,
+    data['options'] = {
+      timeout: options.timeout || 300000,
+      weight: options.weight || 1,
+      group: options.group || null,
+    }
+
+    const jobOptions: JobsOptions = {
       priority: options.priority,
       attempts: options.maxRetries,
       backoff: {
         type: "exponential",
         delay: options.retryDelay || 3000,
       },
-      timeout: options.timeout || 300000, // 5 minutes
       removeOnComplete: options.removeOnComplete || false,
       repeat: options.schedule,
       jobId: task.id,
+      removeOnFail: options.removeOnFail || false,
+      removeDependencyOnFailure: options.removeDependencyOnFailure || false,
+      // @ts-ignore
+      weight: options.weight || 1,
+      // @ts-ignore
+      group: options.group || null,
     };
 
     const job = jobOptions.repeat?.pattern
-      ? await queue!.upsertJobScheduler(jobOptions.id, jobOptions.repeat, {
+      ? await queue!.upsertJobScheduler(task.id, jobOptions.repeat, {
         name,
         data,
         opts: jobOptions,
@@ -573,7 +581,8 @@ export class QueueManager {
 
     const group = await this.getGroup(groupName);
     await group.addTask(methodName, taskOptions, taskData);
-    await group.startProcessing();
+    group.startProcessing();
+    return;
   }
 
   async ensureTaskInQueue(task: Task, queueName: string): Promise<void> {

packages/core/src/types/interfaces.ts

@@ -33,6 +33,10 @@ export interface TaskOptions {
   removeOnComplete?: boolean;
   /** Weight of the task within its group (higher weight = higher priority) */
   weight?: number;
+  /** Remove the task from the queue after failure */
+  removeOnFail?: boolean;
+  /** Remove the task from the queue after failure */
+  removeDependencyOnFailure?: boolean;
 }
 
 /**

packages/core/src/workers/index.ts

@@ -94,6 +94,9 @@ export class Worker extends BullWorker {
     const startTime = Date.now();
     const group = job.data.options?.group;
 
+    // @ts-ignore
+    const timeout = job.data.options?.timeout || job.opts['timeout'] || 300000; // in milliseconds, defaults to 5 minutes
+
     // Track active task
     await this.addActiveTask(job.id!, job.name);
 
@@ -110,6 +113,7 @@ export class Worker extends BullWorker {
         name: job.name,
         group: group,
         data: job.data.args ?? job.data,
+        timeout: timeout
       });
 
       let data = job.data.args ?? job.data;
@@ -125,7 +129,6 @@ export class Worker extends BullWorker {
 
       await job.updateProgress(0);
 
-      const timeout = job.data.options?.timeout || 300000; // in milliseconds, defaults to 5 minutes
       const timeoutPromise = new Promise((_, reject) => {
         setTimeout(() => {
           reject(new Error(`Task ${job.name} timed out after ${timeout}ms`));