Add tests, and new version of getJobs()

Author: adam-fowler

Sources/JobsPostgres/PostgresJobsQueue+cleanup.swift

@@ -111,7 +111,7 @@ extension PostgresJobQueue {
             /// wait for migrations to complete before running job queue cleanup
             try await self.migrations.waitUntilCompleted()
             _ = try await self.client.withTransaction(logger: logger) { connection in
-                self.logger.info("Update Jobs")
+                self.logger.info("Cleanup Jobs")
                 try await self.updateJobsOnInit(withStatus: .pending, onInit: pendingJobs, connection: connection)
                 try await self.updateJobsOnInit(withStatus: .processing, onInit: processingJobs, connection: connection)
                 try await self.updateJobsOnInit(withStatus: .failed, onInit: failedJobs, connection: connection)
@@ -148,7 +148,7 @@ extension PostgresJobQueue {
             )
 
         case .rerun:
-            let jobs = try await getJobs(withStatus: status)
+            let jobs = try await getJobs(withStatus: status, connection: connection)
             self.logger.info("Moving \(jobs.count) jobs with status: \(status) to job queue")
             for jobID in jobs {
                 try await self.addToQueue(jobID: jobID, queueName: configuration.queueName, options: .init(), connection: connection)

Sources/JobsPostgres/PostgresJobsQueue.swift

@@ -502,6 +502,23 @@ public final class PostgresJobQueue: JobQueueDriver, CancellableJobQueue, Resuma
         return jobs
     }
 
+    func getJobs(withStatus status: Status, connection: PostgresConnection) async throws -> [JobID] {
+        let stream = try await connection.query(
+            """
+            SELECT
+                id
+            FROM swift_jobs.jobs
+            WHERE status = \(status) AND queue_name = \(configuration.queueName)
+            """,
+            logger: self.logger
+        )
+        var jobs: [JobID] = []
+        for try await id in stream.decode(JobID.self, context: .default) {
+            jobs.append(id)
+        }
+        return jobs
+    }
+
     let jobRegistry: JobRegistry
 }
 

Tests/JobsPostgresTests/JobsTests.swift

@@ -875,6 +875,8 @@ final class JobsTests: XCTestCase {
 
             let completedJobs = try await jobQueue.queue.getJobs(withStatus: .completed)
             XCTAssertEqual(completedJobs.count, 3)
+            try await jobQueue.queue.cleanup(completedJobs: .remove(maxAge: .seconds(10)))
+            XCTAssertEqual(completedJobs.count, 3)
             try await jobQueue.queue.cleanup(completedJobs: .remove(maxAge: .seconds(0)))
             let zeroJobs = try await jobQueue.queue.getJobs(withStatus: .completed)
             XCTAssertEqual(zeroJobs.count, 0)
@@ -912,6 +914,36 @@ final class JobsTests: XCTestCase {
         }
     }
 
+    func testCleanupProcessingJobs() async throws {
+        let jobQueue = try await self.createJobQueue(numWorkers: 1)
+        let jobName = JobName<Int>("testCancelledJobRetention")
+        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 job = try await jobQueue.queue.popFirst()
+            XCTAssertEqual(jobID, job?.id)
+            _ = try await jobQueue.push(jobName, parameters: 1)
+            _ = try await jobQueue.queue.popFirst()
+
+            var processingJobs = try await jobQueue.queue.getJobs(withStatus: .processing)
+            XCTAssertEqual(processingJobs.count, 2)
+
+            try await jobQueue.queue.cleanup(processingJobs: .remove)
+
+            processingJobs = try await jobQueue.queue.getJobs(withStatus: .processing)
+            XCTAssertEqual(processingJobs.count, 0)
+
+            group.cancelAll()
+        }
+    }
+
     func testCleanupJob() async throws {
         try await self.testJobQueue(
             numWorkers: 1,