Add cleanup of paused jobs (#44)

* Add cleanup of paused jobs

* Update comment

Author: adam-fowler

Sources/JobsPostgres/PostgresJobsQueue+cleanup.swift

@@ -24,20 +24,24 @@ public struct PostgresJobCleanupParameters: Sendable & Codable {
     let completedJobs: PostgresJobQueue.JobCleanup
     let failedJobs: PostgresJobQueue.JobCleanup
     let cancelledJobs: PostgresJobQueue.JobCleanup
+    let pausedJobs: PostgresJobQueue.JobCleanup
 
     ///  Initialize PostgresJobCleanupParameters
     /// - Parameters:
     ///   - completedJobs: What to do with completed jobs
     ///   - failedJobs: What to do with failed jobs
     ///   - cancelledJobs: What to do with cancelled jobs
+    ///   - pausedJobs: What to do with paused jobs
     public init(
         completedJobs: PostgresJobQueue.JobCleanup = .doNothing,
         failedJobs: PostgresJobQueue.JobCleanup = .doNothing,
-        cancelledJobs: PostgresJobQueue.JobCleanup = .doNothing
+        cancelledJobs: PostgresJobQueue.JobCleanup = .doNothing,
+        pausedJobs: PostgresJobQueue.JobCleanup = .doNothing
     ) {
         self.failedJobs = failedJobs
         self.completedJobs = completedJobs
         self.cancelledJobs = cancelledJobs
+        self.pausedJobs = pausedJobs
     }
 }
 
@@ -51,9 +55,13 @@ extension PostgresJobQueue {
         }
         let rawValue: RawValue
 
+        /// Do nothing to jobs
         public static var doNothing: Self { .init(rawValue: .doNothing) }
+        /// Re-queue jobs for running again
         public static var rerun: Self { .init(rawValue: .rerun) }
+        /// Delete jobs
         public static var remove: Self { .init(rawValue: .remove(maxAge: nil)) }
+        /// Delete jobs that are older than maxAge
         public static func remove(maxAge: Duration) -> Self { .init(rawValue: .remove(maxAge: maxAge)) }
     }
 
@@ -75,6 +83,7 @@ extension PostgresJobQueue {
                     completedJobs: parameters.completedJobs,
                     failedJobs: parameters.failedJobs,
                     cancelledJobs: parameters.cancelledJobs,
+                    pausedJobs: parameters.pausedJobs,
                     logger: self.logger
                 )
             }
@@ -83,23 +92,23 @@ extension PostgresJobQueue {
 
     ///  Cleanup job queues
     ///
-    /// This function is used to re-run or delete jobs in a certain state. Failed jobs can be
-    /// pushed back into the pending queue to be re-run or removed. When called at startup in
-    /// theory no job should be set to processing, or set to pending but not in the queue. but if
-    /// your job server crashes these states are possible, so we also provide options to re-queue
-    /// these jobs so they are run again.
+    /// This function is used to re-run or delete jobs in a certain state. Failed, completed,
+    /// cancelled and paused jobs can be pushed back into the pending queue to be re-run or removed.
+    /// When called at startup in theory no job should be set to processing, or set to pending but
+    /// not in the queue. but if your job server crashes these states are possible, so we also provide
+    /// options to re-queue these jobs so they are run again.
     ///
-    /// The job queue needs to be running when you call cleanup. You can call `cleanup` with
-    /// `failedJobs`` set to whatever you like at any point to re-queue failed jobs. Moving processing
-    /// or pending jobs should only be done if you are certain there is nothing else processing
-    /// the job queue.
+    /// You can call `cleanup` with `failedJobs`, `completedJobs`, `cancelledJobs` or `pausedJobs` set
+    /// to whatever you like at any point to re-queue failed jobs. Moving processing or pending jobs
+    /// should only be done if you are certain there is nothing processing the job queue.
     ///
     /// - Parameters:
     ///   - pendingJobs: What to do with jobs tagged as pending
     ///   - processingJobs: What to do with jobs tagged as processing
     ///   - completedJobs: What to do with jobs tagged as completed
     ///   - failedJobs: What to do with jobs tagged as failed
     ///   - cancelledJobs: What to do with jobs tagged as cancelled
+    ///   - pausedJobs: What to do with jobs tagged as paused
     ///   - logger: Optional logger to use when performing cleanup
     /// - Throws:
     public func cleanup(
@@ -108,6 +117,7 @@ extension PostgresJobQueue {
         completedJobs: JobCleanup = .doNothing,
         failedJobs: JobCleanup = .doNothing,
         cancelledJobs: JobCleanup = .doNothing,
+        pausedJobs: JobCleanup = .doNothing,
         logger: Logger? = nil
     ) async throws {
         let logger = logger ?? self.logger
@@ -121,6 +131,7 @@ extension PostgresJobQueue {
                 try await self.updateJobsOnInit(withStatus: .failed, onInit: failedJobs, connection: connection)
                 try await self.updateJobsOnInit(withStatus: .completed, onInit: completedJobs, connection: connection)
                 try await self.updateJobsOnInit(withStatus: .cancelled, onInit: cancelledJobs, connection: connection)
+                try await self.updateJobsOnInit(withStatus: .paused, onInit: pausedJobs, connection: connection)
             }
         } catch let error as PSQLError {
             logger.error(

Tests/JobsPostgresTests/JobsTests.swift

@@ -880,6 +880,36 @@ struct JobsTests {
         }
     }
 
+    @Test func testPausedJobRetention() async throws {
+        let jobQueue = try await self.createJobQueue(
+            configuration: .init(queueName: "testPausedJobRetention")
+        )
+        let jobName = JobName<Int>("testPausedJobRetention")
+        jobQueue.registerJob(name: jobName) { _, _ in }
+
+        try await withThrowingTaskGroup(of: Void.self) { group in
+            group.addTask {
+                // run postgres client
+                await jobQueue.queue.client.run()
+            }
+            try await jobQueue.queue.migrations.apply(client: jobQueue.queue.client, logger: jobQueue.logger, dryRun: false)
+
+            let jobId = try await jobQueue.push(jobName, parameters: 1)
+            let jobId2 = try await jobQueue.push(jobName, parameters: 2)
+
+            try await jobQueue.pauseJob(jobID: jobId)
+            try await jobQueue.pauseJob(jobID: jobId2)
+
+            var cancelledJobs = try await jobQueue.queue.getJobs(withStatus: .paused)
+            #expect(cancelledJobs.count == 2)
+            try await jobQueue.queue.cleanup(pausedJobs: .remove(maxAge: .seconds(0)))
+            cancelledJobs = try await jobQueue.queue.getJobs(withStatus: .paused)
+            #expect(cancelledJobs.count == 0)
+
+            group.cancelAll()
+        }
+    }
+
     @Test func testCleanupProcessingJobs() async throws {
         let jobQueue = try await self.createJobQueue()
         let jobName = JobName<Int>("testCleanupProcessingJobs")