Create separate cleanup job for each queue

Author: adam-fowler

Package.swift

@@ -10,7 +10,7 @@ let package = Package(
     ],
     dependencies: [
         // TODO: use a released version of swift-jobs
-        .package(url: "https://github.com/hummingbird-project/swift-jobs.git", branch: "main"),
+        .package(url: "https://github.com/hummingbird-project/swift-jobs.git", branch: "jobschedule-addjob"),
         .package(url: "https://github.com/hummingbird-project/postgres-migrations.git", from: "0.1.0"),
         .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.25.0"),
     ],

Sources/JobsPostgres/JobCleanupParameters.swift

@@ -20,7 +20,48 @@ import NIOCore
 import PostgresMigrations
 import PostgresNIO
 
+/// Parameters for Cleanup job
+public struct JobCleanupParameters: Sendable & Codable {
+    let failedJobs: PostgresJobQueue.JobCleanup
+    let completedJobs: PostgresJobQueue.JobCleanup
+    let cancelledJobs: PostgresJobQueue.JobCleanup
+
+    public init(
+        failedJobs: PostgresJobQueue.JobCleanup = .doNothing,
+        completedJobs: PostgresJobQueue.JobCleanup = .doNothing,
+        cancelledJobs: PostgresJobQueue.JobCleanup = .doNothing
+    ) {
+        self.failedJobs = failedJobs
+        self.completedJobs = completedJobs
+        self.cancelledJobs = cancelledJobs
+    }
+}
+
 extension PostgresJobQueue {
+    /// clean up job name.
+    ///
+    /// Use this with the ``JobSchedule`` to schedule a cleanup of
+    /// failed, cancelled or completed jobs
+    public var cleanupJob: JobName<JobCleanupParameters> {
+        .init("_JobCleanup_\(self.configuration.queueName)")
+    }
+
+    /// register clean up job on queue
+    func registerCleanupJob() {
+        self.registerJob(
+            JobDefinition(name: cleanupJob, parameters: JobCleanupParameters.self, retryStrategy: .dontRetry) { parameters, context in
+                try await self.cleanup(
+                    failedJobs: parameters.failedJobs,
+                    processingJobs: .doNothing,
+                    pendingJobs: .doNothing,
+                    completedJobs: parameters.completedJobs,
+                    cancelledJobs: parameters.cancelledJobs,
+                    logger: self.logger
+                )
+            }
+        )
+    }
+
     ///  Cleanup job queues
     ///
     /// This function is used to re-run or delete jobs in a certain state. Failed jobs can be
@@ -80,7 +121,7 @@ extension PostgresJobQueue {
                 if let olderThan {
                     .now - Double(olderThan.components.seconds)
                 } else {
-                    .distantPast
+                    .distantFuture
                 }
             try await connection.query(
                 """

Sources/JobsPostgres/PostgresJobsQueue+cleanup.swift

@@ -165,7 +165,6 @@ public final class PostgresJobQueue: JobQueueDriver, CancellableJobQueue, Resuma
     public let configuration: Configuration
     /// Logger used by queue
     public let logger: Logger
-
     let migrations: DatabaseMigrations
     let isStopped: NIOLockedValueBox<Bool>
 
@@ -178,18 +177,7 @@ public final class PostgresJobQueue: JobQueueDriver, CancellableJobQueue, Resuma
         self.isStopped = .init(false)
         self.migrations = migrations
         await migrations.add(CreateSwiftJobsMigrations(), skipDuplicates: true)
-        self.registerJob(
-            JobDefinition(parameters: JobCleanupParameters.self, retryStrategy: .dontRetry) { parameters, context in
-                try await self.cleanup(
-                    failedJobs: parameters.failedJobs,
-                    processingJobs: .doNothing,
-                    pendingJobs: .doNothing,
-                    completedJobs: parameters.completedJobs,
-                    cancelledJobs: parameters.cancelledJobs,
-                    logger: logger
-                )
-            }
-        )
+        self.registerCleanupJob()
     }
 
     public func onInit() async throws {

Sources/JobsPostgres/PostgresJobsQueue.swift

@@ -42,7 +42,7 @@ public struct RetentionPolicy: Sendable {
 
     public init(
         cancelled: RetainData = .retain,
-        completed: RetainData = .retain,
+        completed: RetainData = .never,
         failed: RetainData = .retain
     ) {
         self.cancelled = cancelled

Sources/JobsPostgres/RetentionPolicy.swift

@@ -886,4 +886,67 @@ final class JobsTests: XCTestCase {
             XCTAssertEqual(zeroJobs.count, 0)
         }
     }
+
+    func testCleanupJob() async throws {
+        try await self.testJobQueue(
+            numWorkers: 1,
+            configuration: .init(
+                retentionPolicy: .init(
+                    cancelled: .retain,
+                    completed: .never,
+                    failed: .retain
+                )
+            )
+        ) { jobQueue in
+            try await self.testJobQueue(
+                numWorkers: 1,
+                configuration: .init(
+                    queueName: "SecondQueue",
+                    retentionPolicy: .init(
+                        cancelled: .retain,
+                        completed: .never,
+                        failed: .retain
+                    )
+                )
+            ) { jobQueue2 in
+                let (stream, cont) = AsyncStream.makeStream(of: Void.self)
+                var iterator = stream.makeAsyncIterator()
+                struct TempJob: Sendable & Codable {}
+                let barrierJobName = JobName<TempJob>("barrier")
+                jobQueue.registerJob(name: "testCleanupJob", parameters: String.self) { parameters, context in
+                    throw CancellationError()
+                }
+                jobQueue.registerJob(name: barrierJobName, parameters: TempJob.self) { parameters, context in
+                    cont.yield()
+                }
+                jobQueue2.registerJob(name: "testCleanupJob", parameters: String.self) { parameters, context in
+                    throw CancellationError()
+                }
+                jobQueue2.registerJob(name: barrierJobName, parameters: TempJob.self) { parameters, context in
+                    cont.yield()
+                }
+                try await jobQueue.push("testCleanupJob", parameters: "1")
+                try await jobQueue.push("testCleanupJob", parameters: "2")
+                try await jobQueue.push("testCleanupJob", parameters: "3")
+                try await jobQueue.push(barrierJobName, parameters: .init())
+                try await jobQueue2.push("testCleanupJob", parameters: "1")
+                try await jobQueue2.push(barrierJobName, parameters: .init())
+
+                await iterator.next()
+                await iterator.next()
+
+                let failedJob = try await jobQueue.queue.getJobs(withStatus: .failed)
+                XCTAssertEqual(failedJob.count, 3)
+                try await jobQueue.push(jobQueue.queue.cleanupJob, parameters: .init(failedJobs: .remove))
+                try await jobQueue.push(barrierJobName, parameters: .init())
+
+                await iterator.next()
+
+                let zeroJobs = try await jobQueue.queue.getJobs(withStatus: .failed)
+                XCTAssertEqual(zeroJobs.count, 0)
+                let jobCount2 = try await jobQueue2.queue.getJobs(withStatus: .failed)
+                XCTAssertEqual(jobCount2.count, 1)
+            }
+        }
+    }
 }