PR comments

Author: thoven87

Sources/JobsPostgres/PostgresJobsQueue.swift

@@ -142,17 +142,21 @@ public final class PostgresJobQueue: JobQueueDriver, CancellableJobQueue, Resuma
         let pollTime: Duration
         /// Which Queue to push jobs into
         let queueName: String
+        /// What to do with cancelled jobs
+        let deleteCancelledJobs: Bool
 
         ///  Initialize configuration
         /// - Parameters
         ///   - pollTime: Queue poll time to wait if queue empties
         ///   - queueName: Name of queue we are handing
         public init(
             pollTime: Duration = .milliseconds(100),
-            queueName: String = "default"
+            queueName: String = "default",
+            deleteCancelledJobs: Bool = true
         ) {
             self.pollTime = pollTime
             self.queueName = queueName
+            self.deleteCancelledJobs = deleteCancelledJobs
         }
     }
 
@@ -194,6 +198,10 @@ public final class PostgresJobQueue: JobQueueDriver, CancellableJobQueue, Resuma
     public func cancel(jobID: JobID) async throws {
         try await self.client.withTransaction(logger: logger) { connection in
             try await deleteFromQueue(jobID: jobID, connection: connection)
+            if configuration.deleteCancelledJobs {
+                try await delete(jobID: jobID)
+                return
+            }
             try await setStatus(jobID: jobID, status: .cancelled, connection: connection)
         }
     }

Tests/JobsPostgresTests/JobsTests.swift

@@ -704,22 +704,22 @@ final class JobsTests: XCTestCase {
                 expectation.fulfill()
             }
 
-            let firstJob = try await queue.push(
+            let resumableJob = try await queue.push(
                 TestParameters(value: 20),
                 options: .init(
                     priority: .lowest()
                 )
             )
 
-            try await jobQueue.pauseJob(jobID: firstJob)
-
             try await queue.push(
                 TestParameters(value: 2025),
                 options: .init(
                     priority: .normal()
                 )
             )
 
+            try await jobQueue.pauseJob(jobID: resumableJob)
+
             try await withThrowingTaskGroup(of: Void.self) { group in
                 let serviceGroup = ServiceGroup(services: [queue], logger: queue.logger)
 
@@ -729,16 +729,21 @@ final class JobsTests: XCTestCase {
                 group.addTask {
                     try await serviceGroup.run()
                 }
-
+                
+                let processingJobCount = try await jobQueue.queue.getJobs(withStatus: .processing)
+                // Job 2 has been processed
+                XCTAssertEqual(processingJobCount.count, 0)
+                // Job 1 has not been processed
                 let pausedJobs = try await jobQueue.queue.getJobs(withStatus: .paused)
                 XCTAssertEqual(pausedJobs.count, 1)
-
-                try await jobQueue.resumeJob(jobID: firstJob)
+                // resume job 1
+                try await jobQueue.resumeJob(jobID: resumableJob)
 
                 await fulfillment(of: [expectation], timeout: 10)
                 await serviceGroup.triggerGracefulShutdown()
             }
         }
+        // verify job run order
         XCTAssertEqual(jobExecutionSequence.withLockedValue { $0 }, [2025, 20])
     }