refactor: update job completion and failure handling in queue system

- Renamed `release` method to `completeJob` for clarity across queue implementations.
- Introduced `failJob` method to handle job failures appropriately.
- Updated error handling logic to return false when a job fails.
- Enhanced tests to verify correct behavior for job completion and failure scenarios across different queue drivers.

Author: muniter

src/core/queue.ts

@@ -264,9 +264,11 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
           // Don't let plugin errors affect job completion
         }
 
-        // Complete the job if successful
+        // Complete the job if successful, otherwise mark as failed
         if (success) {
-          await this.release(message);
+          await this.completeJob(message);
+        } else {
+          await this.failJob(message, jobError || new Error('Unknown job failure'));
         }
       }
     } finally {
@@ -320,7 +322,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    * 
    * @param message - The queue message that failed to process
    * @param error - The error that occurred during processing
-   * @returns Promise resolving to true (job is considered handled despite the error)
+   * @returns Promise resolving to false (job failed)
    * @protected
    */
   protected async handleError(message: QueueMessage, error: unknown): Promise<boolean> {
@@ -336,7 +338,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
       this.emit('afterError', errorEvent);
     }
 
-    return true; // Job is considered handled (failed)
+    return false; // Job failed
   }
 
   protected async sleep(ms: number): Promise<void> {
@@ -367,14 +369,25 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
   protected abstract reserve(timeout: number): Promise<QueueMessage | null>;
 
   /**
-   * Releases a processed job from the queue (marks as complete).
+   * Marks a job as successfully completed and removes it from the queue.
    * 
-   * @param message - The queue message to release
-   * @returns Promise that resolves when job is released
+   * @param message - The queue message to complete
+   * @returns Promise that resolves when job is marked as complete
    * @protected
    * @abstract
    */
-  protected abstract release(message: QueueMessage): Promise<void>;
+  protected abstract completeJob(message: QueueMessage): Promise<void>;
+  
+  /**
+   * Marks a job as failed and handles failure appropriately (remove or retry).
+   * 
+   * @param message - The queue message that failed
+   * @param error - The error that caused the failure
+   * @returns Promise that resolves when job failure is handled
+   * @protected
+   * @abstract
+   */
+  protected abstract failJob(message: QueueMessage, error: unknown): Promise<void>;
 
   /**
    * Retrieves the current status of a job by its ID.

src/drivers/db.ts

@@ -33,10 +33,15 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
     };
   }
 
-  protected async release(message: QueueMessage): Promise<void> {
+  protected async completeJob(message: QueueMessage): Promise<void> {
     await this.db.completeJob(message.id);
   }
 
+  protected async failJob(message: QueueMessage, error: unknown): Promise<void> {
+    const errorMessage = error instanceof Error ? error.message : String(error);
+    await this.db.failJob(message.id, errorMessage);
+  }
+
   async status(id: string): Promise<JobStatus> {
     const status = await this.db.getJobStatus(id);
     return status || 'done';

src/drivers/file.ts

@@ -148,7 +148,13 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
     }
   }
 
-  protected async release(message: QueueMessage): Promise<void> {
+  protected async completeJob(message: QueueMessage): Promise<void> {
+    await this.complete(message.id);
+  }
+
+  protected async failJob(message: QueueMessage, error: unknown): Promise<void> {
+    // For file queue, we'll remove failed jobs too (similar to complete)
+    // Future enhancement could track failed jobs in the index
     await this.complete(message.id);
   }
 

src/drivers/redis.ts

@@ -216,7 +216,7 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     };
   }
 
-  protected async release(message: QueueMessage): Promise<void> {
+  protected async completeJob(message: QueueMessage): Promise<void> {
     // Remove from reserved queue
     await this.redis.zrem(this.reservedKey, message.id);
 
@@ -225,6 +225,15 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     await this.redis.hdel(this.attemptsKey, message.id);
   }
 
+  protected async failJob(message: QueueMessage, error: unknown): Promise<void> {
+    // Remove from reserved queue
+    await this.redis.zrem(this.reservedKey, message.id);
+    
+    // Remove job data (Redis doesn't track failed job history by default)
+    await this.redis.hdel(this.messagesKey, message.id);
+    await this.redis.hdel(this.attemptsKey, message.id);
+  }
+
   async status(id: string): Promise<JobStatus> {
     // Check if job data exists
     const exists = await this.redis.hget(this.messagesKey, id);

src/drivers/sqs.ts

@@ -22,14 +22,19 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
   TJobMap,
   SqsJobRequest<any>
 > {
+  #onFailure: "delete" | "leaveInQueue"
+
   constructor(
     private client: SQSClient,
     private queueUrl: string,
-    options: QueueOptions = {}
+    options: QueueOptions & { onFailure: "delete" | "leaveInQueue" } = {
+      onFailure: "delete",
+    }
   ) {
     super(options);
     // SQS supports long polling via WaitTimeSeconds
     this.supportsLongPolling = true;
+    this.#onFailure = options.onFailure;
   }
 
   protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
@@ -57,7 +62,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
       DelaySeconds: meta.delay || 0,
       MessageAttributes: messageAttributes,
     });
-    
+
     const result = await this.client.send(command);
 
     if (!result.MessageId) {
@@ -73,7 +78,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
       WaitTimeSeconds: timeout,
       MessageAttributeNames: ["All"],
     });
-    
+
     const result = await this.client.send(command);
 
     if (!result.Messages || result.Messages.length === 0) {
@@ -105,7 +110,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
         ReceiptHandle: message.ReceiptHandle,
         VisibilityTimeout: meta.ttr,
       });
-      
+
       await this.client.send(visibilityCommand);
     }
 
@@ -119,21 +124,43 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
     };
   }
 
-  protected async release(message: QueueMessage): Promise<void> {
+  protected async completeJob(message: QueueMessage): Promise<void> {
     if (!message.meta.receiptHandle) {
       throw new Error(
-        "Cannot release SQS message: receiptHandle is missing from metadata"
+        "Cannot complete SQS message: receiptHandle is missing from metadata"
       );
     }
 
     const deleteCommand = new DeleteMessageCommand({
       QueueUrl: this.queueUrl,
       ReceiptHandle: message.meta.receiptHandle,
     });
-    
+
     await this.client.send(deleteCommand);
   }
 
+  protected async failJob(
+    message: QueueMessage,
+    error: unknown
+  ): Promise<void> {
+    if (this.#onFailure === "leaveInQueue") {
+      if (!message.meta.receiptHandle) {
+        throw new Error(
+          "Cannot fail SQS message: receiptHandle is missing from metadata"
+        );
+      }
+
+      // For SQS, we delete failed messages too since SQS doesn't track failure states
+      // Future enhancement could send to a dead letter queue
+      const deleteCommand = new DeleteMessageCommand({
+        QueueUrl: this.queueUrl,
+        ReceiptHandle: message.meta.receiptHandle,
+      });
+
+      await this.client.send(deleteCommand);
+    }
+  }
+
   async status(id: string): Promise<JobStatus> {
     throw new Error("SQS does not support status");
   }