feat(retry backoff): Adding retry with backoff (#6)

* feat(retry backoff): Adding retry with backoff * dropped retry func * use main branch
hummingbird-project · Sep 14, 2024 · a455c8f · a455c8f
1 parent 34e144d
commit a455c8f
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 5 deletions.
diff --git a/Sources/JobsRedis/RedisJobQueue.swift b/Sources/JobsRedis/RedisJobQueue.swift
@@ -114,13 +114,15 @@ public final class RedisJobQueue: JobQueueDriver {
     /// - Returns: Job ID
     @discardableResult public func push(_ buffer: ByteBuffer, options: JobOptions) async throws -> JobID {
         let jobInstanceID = JobID(delayUntil: options.delayUntil)
-
-        try await self.set(jobId: jobInstanceID, buffer: buffer)
-
-        _ = try await self.redisConnectionPool.wrappedValue.lpush(jobInstanceID.redisKey, into: self.configuration.queueKey).get()
+        try await self.addToQueue(jobInstanceID, buffer: buffer)
         return jobInstanceID
     }
 
+    private func addToQueue(_ jobId: JobID, buffer: ByteBuffer) async throws {
+        try await self.set(jobId: jobId, buffer: buffer)
+        _ = try await self.redisConnectionPool.wrappedValue.lpush(jobId.redisKey, into: self.configuration.queueKey).get()
+    }
+
     /// Flag job is done
     ///
     /// Removes  job id from processing queue

diff --git a/Tests/JobsRedisTests/RedisJobsTests.swift b/Tests/JobsRedisTests/RedisJobsTests.swift
@@ -74,7 +74,12 @@ final class RedisJobsTests: XCTestCase {
                 )
             ),
             numWorkers: numWorkers,
-            logger: logger
+            logger: logger,
+            options: .init(
+                maximumBackoff: 0.01,
+                maxJitter: 0.01,
+                minJitter: 0.0
+            )
         )
 
         return try await withThrowingTaskGroup(of: Void.self) { group in
@@ -176,6 +181,40 @@ final class RedisJobsTests: XCTestCase {
         }
     }
 
+    func testErrorRetryAndThenSucceed() async throws {
+        let jobIdentifer = JobIdentifier<Int>(#function)
+        let expectation = XCTestExpectation(description: "TestJob.execute was called", expectedFulfillmentCount: 2)
+        let currentJobTryCount: NIOLockedValueBox<Int> = .init(0)
+        struct FailedError: Error {}
+        try await self.testJobQueue(numWorkers: 1) { jobQueue in
+            jobQueue.registerJob(id: jobIdentifer, maxRetryCount: 3) { _, _ in
+                defer {
+                    currentJobTryCount.withLockedValue {
+                        $0 += 1
+                    }
+                }
+                expectation.fulfill()
+                if currentJobTryCount.withLockedValue({ $0 }) == 0 {
+                    throw FailedError()
+                }
+            }
+            try await jobQueue.push(id: jobIdentifer, parameters: 0)
+
+            await self.fulfillment(of: [expectation], timeout: 5)
+            try await Task.sleep(for: .milliseconds(200))
+
+            let failedJobs = try await jobQueue.queue.redisConnectionPool.wrappedValue.llen(of: jobQueue.queue.configuration.failedQueueKey).get()
+            XCTAssertEqual(failedJobs, 0)
+
+            let pendingJobs = try await jobQueue.queue.redisConnectionPool.wrappedValue.llen(of: jobQueue.queue.configuration.queueKey).get()
+            XCTAssertEqual(pendingJobs, 0)
+
+            let processingJobs = try await jobQueue.queue.redisConnectionPool.wrappedValue.llen(of: jobQueue.queue.configuration.processingQueueKey).get()
+            XCTAssertEqual(processingJobs, 0)
+        }
+        XCTAssertEqual(currentJobTryCount.withLockedValue { $0 }, 2)
+    }
+
     func testJobSerialization() async throws {
         struct TestJobParameters: Codable {
             let id: Int