Simplify job query and allow job rerun on start if specified

Author: thoven87

Package.swift

@@ -9,7 +9,7 @@ let package = Package(
         .library(name: "JobsPostgres", targets: ["JobsPostgres"])
     ],
     dependencies: [
-        .package(url: "https://github.com/hummingbird-project/swift-jobs.git", from: "1.0.0-beta.4"),
+        .package(url: "https://github.com/hummingbird-project/swift-jobs.git", from: "1.0.0-beta.7"),
         .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.21.0"),
     ],

Sources/JobsPostgres/PostgresJobsQueue.swift

@@ -130,7 +130,9 @@ public final class PostgresJobQueue: JobQueueDriver {
                 try await self.updateJobsOnInit(withStatus: .failed, onInit: self.configuration.failedJobsInitialization, connection: connection)
             }
         } catch let error as PSQLError {
-            print("\(String(reflecting: error))")
+            logger.error("JobQueue initialization failed", metadata: [
+                "error": "\(String(reflecting: error))"
+            ])
             throw error
         }
     }
@@ -194,17 +196,19 @@ public final class PostgresJobQueue: JobQueueDriver {
 
                     let stream = try await connection.query(
                         """
-                        DELETE FROM
-                            _hb_pg_job_queue
-                        USING (
-                            SELECT job_id FROM _hb_pg_job_queue
+                        WITH next_job AS (
+                            SELECT
+                                job_id
+                            FROM _hb_pg_job_queue
                             WHERE (delayed_until IS NULL OR delayed_until <= NOW())
                             ORDER BY createdAt, delayed_until ASC
-                            LIMIT 1
                             FOR UPDATE SKIP LOCKED
-                        ) queued
-                        WHERE queued.job_id = _hb_pg_job_queue.job_id
-                        RETURNING _hb_pg_job_queue.job_id
+                            LIMIT 1
+                        )
+                        DELETE FROM
+                            _hb_pg_job_queue
+                        WHERE job_id = (SELECT job_id FROM next_job)
+                        RETURNING job_id
                         """,
                         logger: self.logger
                     )
@@ -214,7 +218,7 @@ public final class PostgresJobQueue: JobQueueDriver {
                     }
                     // select job from job table
                     let stream2 = try await connection.query(
-                        "SELECT job FROM _hb_pg_jobs WHERE id = \(jobId) FOR UPDATE SKIP LOCKED",
+                        "SELECT job FROM _hb_pg_jobs WHERE id = \(jobId)",
                         logger: self.logger
                     )
 
@@ -274,7 +278,9 @@ public final class PostgresJobQueue: JobQueueDriver {
             INSERT INTO _hb_pg_job_queue (job_id, createdAt, delayed_until)
             VALUES (\(jobId), \(Date.now), \(delayUntil))
             ON CONFLICT (job_id)
-            DO UPDATE SET delayed_until = \(delayUntil)
+            DO UPDATE
+            SET delayed_until = COALESCE(_hb_pg_job_queue.delayed_until, EXCLUDED.delayed_until),
+                createdAt = \(Date.now)
             """,
             logger: self.logger
         )
@@ -315,8 +321,6 @@ public final class PostgresJobQueue: JobQueueDriver {
             )
 
         case .rerun:
-            guard status != .pending else { return }
-
             let jobs = try await getJobs(withStatus: status)
             self.logger.info("Moving \(jobs.count) jobs with status: \(status) to job queue")
             for jobId in jobs {