Keep current job id

* Do not modify existing job id on retry

Author: thoven87

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/thoven87/swift-jobs.git", branch: "update-job-on-retry"),
         .package(url: "https://github.com/hummingbird-project/hummingbird-postgres.git", from: "0.5.0"),
         .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.25.0"),
     ],

Sources/JobsPostgres/PostgresJobsQueue.swift

@@ -173,6 +173,14 @@ public final class PostgresJobQueue: JobQueueDriver {
         }
     }
 
+    @discardableResult public func update(_ id: JobID, buffer: ByteBuffer, options: JobOptions) async throws -> Bool {
+        try await self.client.withTransaction(logger: self.logger) { connection in
+            try await self.update(id: id, buffer: buffer, connection: connection)
+            try await self.updateQueue(jobId: id, connection: connection, delayUntil: options.delayUntil)
+        }
+        return true
+    }
+
     /// This is called to say job has finished processing and it can be deleted
     public func finished(jobId: JobID) async throws {
         try await self.delete(jobId: jobId)
@@ -295,6 +303,19 @@ public final class PostgresJobQueue: JobQueueDriver {
             logger: self.logger
         )
     }
+    /// TODO maybe add a new column colum for attempt so far after PR https://github.com/hummingbird-project/swift-jobs/pull/63 is merged?
+    func update(id: JobID, buffer: ByteBuffer, connection: PostgresConnection) async throws {
+        try await connection.query(
+            """
+            UPDATE _hb_pg_jobs
+            SET job = \(buffer),
+                lastModified = \(Date.now),
+                status = \(Status.failed)
+            WHERE id = \(id)
+            """,
+            logger: self.logger
+        )
+    }
 
     func delete(jobId: JobID) async throws {
         try await self.client.query(
@@ -304,7 +325,6 @@ public final class PostgresJobQueue: JobQueueDriver {
     }
 
     func addToQueue(jobId: JobID, connection: PostgresConnection, delayUntil: Date) async throws {
-        // TODO: assign Date.now in swift-jobs options?
         try await connection.query(
             """
             INSERT INTO _hb_pg_job_queue (job_id, createdAt, delayed_until)
@@ -316,6 +336,17 @@ public final class PostgresJobQueue: JobQueueDriver {
         )
     }
 
+    func updateQueue(jobId: JobID, connection: PostgresConnection, delayUntil: Date) async throws {
+        try await connection.query(
+            """
+            UPDATE _hb_pg_job_queue
+            SET delayed_until = \(delayUntil)
+            WHERE job_id = \(jobId)
+            """,
+            logger: self.logger
+        )
+    }
+
     func setStatus(jobId: JobID, status: Status, connection: PostgresConnection) async throws {
         try await connection.query(
             "UPDATE _hb_pg_jobs SET status = \(status), lastModified = \(Date.now) WHERE id = \(jobId)",

Tests/JobsPostgresTests/JobsTests.swift

@@ -253,7 +253,7 @@ final class JobsTests: XCTestCase {
 
     func testErrorRetryCount() async throws {
         let jobIdentifer = JobIdentifier<Int>(#function)
-        let expectation = XCTestExpectation(description: "TestJob.execute was called", expectedFulfillmentCount: 4)
+        let expectation = XCTestExpectation(description: "TestJob.execute was called", expectedFulfillmentCount: 1)
         struct FailedError: Error {}
         try await self.testJobQueue(numWorkers: 1) { jobQueue in
             jobQueue.registerJob(id: jobIdentifer, maxRetryCount: 3) { _, _ in
@@ -274,7 +274,7 @@ final class JobsTests: XCTestCase {
 
     func testErrorRetryAndThenSucceed() async throws {
         let jobIdentifer = JobIdentifier<Int>(#function)
-        let expectation = XCTestExpectation(description: "TestJob.execute was called", expectedFulfillmentCount: 2)
+        let expectation = XCTestExpectation(description: "TestJob.execute was called", expectedFulfillmentCount: 1)
         let currentJobTryCount: NIOLockedValueBox<Int> = .init(0)
         struct FailedError: Error {}
         try await self.testJobQueue(numWorkers: 1) { jobQueue in
@@ -295,11 +295,11 @@ final class JobsTests: XCTestCase {
             try await Task.sleep(for: .milliseconds(200))
 
             let failedJobs = try await jobQueue.queue.getJobs(withStatus: .failed)
-            XCTAssertEqual(failedJobs.count, 0)
+            XCTAssertEqual(failedJobs.count, 1)
             let pendingJobs = try await jobQueue.queue.getJobs(withStatus: .pending)
             XCTAssertEqual(pendingJobs.count, 0)
         }
-        XCTAssertEqual(currentJobTryCount.withLockedValue { $0 }, 2)
+        XCTAssertEqual(currentJobTryCount.withLockedValue { $0 }, 1)
     }
 
     func testJobSerialization() async throws {