Reset retry counters on queue:retry

CHANGELOG.md

@@ -1,6 +1,6 @@
 ## Unreleased
 
-- New opt-in job settings: `retryUntil` (retry until a time, not `tries` times), `maxExceptions` (fail after this many tries that threw) and `failOnTimeout` (fail at once on a timeout). Jobs without them work as before. `retryUntil` is read once at dispatch and saved with the job. Rows saved before this change are read as jobs with no `retryUntil`.
+- New opt-in job settings: `retryUntil` (retry until a time, not `tries` times), `maxExceptions` (fail after this many tries that threw) and `failOnTimeout` (fail at once on a timeout). Jobs without them work as before. `retryUntil` is read once at dispatch and saved with the job. Rows saved before this change are read as jobs with no `retryUntil`. `queue:retry` gives a failed job a fresh start: the tries and the exceptions count go back to 0, and `retryUntil` is read again from the job class.
 - New: `dispatch(delayUntil: DateTime)` waits until a time. `delay: Duration` still works. Give one of them, not both.
 - New: `dispatchIf(condition)` and `dispatchUnless(condition)`. They return the job id, or null when the job was not dispatched.
 - New worker flags: `--backoff=SECONDS[,SECONDS...]` (the retry wait of a job with no `backoff`), `--max-jobs=N`, `--max-time=SECONDS` and `--memory=MB` (stop the worker after the running jobs). All are off by default.

lib/src/drivers/database_queue_driver.dart

@@ -161,7 +161,7 @@ class DatabaseQueueDriver extends QueueDriver {
     return db.transaction(() async {
       FailedJob? job = await _find(db, id);
       if (job == null) return false;
-      await push(JobPayload.decode(job.payload).withAttempts(0));
+      await push(JobPayload.decode(job.payload).retried());
       await _forget(db, id);
       return true;
     });

lib/src/drivers/redis_queue_driver.dart

@@ -142,7 +142,7 @@ return 1''';
     String? text = await redis.hget(_failed, id);
     if (text == null) return false;
     FailedJob job = _failedJob(id, text);
-    await push(JobPayload.decode(job.payload).withAttempts(0));
+    await push(JobPayload.decode(job.payload).retried());
     await redis.hdel(_failed, id);
     return true;
   }

lib/src/queue.dart

@@ -202,6 +202,21 @@ class Queue {
     );
   }
 
+  /// `retryUntil` of the job class of [payload], read now, as unix time
+  /// in milliseconds. Null when the job has none, or can not be read
+  /// (the worker then fails it with the real error).
+  static int? retryUntilOf(JobPayload payload) {
+    try {
+      JobCodec? codec = config.jobs.byName(payload.job);
+      if (codec == null) return null;
+      String data = readData(payload.data, payload.job);
+      DoxJob job = codec.decode(jsonDecode(data) as Map<String, Object?>);
+      return job.retryUntil?.millisecondsSinceEpoch;
+    } catch (_) {
+      return null;
+    }
+  }
+
   /// the job data has only JSON types, dates and enums
   static Object? _toJson(Object? value) {
     if (value is DateTime) return value.toIso8601String();

lib/src/queue_driver.dart

@@ -1,6 +1,8 @@
 import 'dart:convert';
 import 'dart:math';
 
+import 'queue.dart';
+
 /// One queued job, as it is saved: the job name, its data and how many
 /// times a worker took it.
 class JobPayload {
@@ -58,6 +60,19 @@ class JobPayload {
     );
   }
 
+  /// A copy for `queue:retry`: a fresh start, like a new dispatch. No
+  /// tries and no exceptions yet. [retryUntil] is read again from the
+  /// job class, so the job gets a new time window.
+  JobPayload retried() {
+    return JobPayload(
+      id: id,
+      job: job,
+      queue: queue,
+      data: data,
+      retryUntil: retryUntil == null ? null : Queue.retryUntilOf(this),
+    );
+  }
+
   /// a copy with other [exceptions]
   JobPayload withExceptions(int exceptions) {
     return JobPayload(

test/job_options_test.dart

@@ -233,6 +233,35 @@ void main() {
         expect(JobLog.failed, <String>['until']);
       });
 
+      test('a retried job gets its full maxExceptions again', () async {
+        await OptionsJob(key: 'again', maxTries: 10, exceptions: 2).dispatch();
+        await workUntilFailed('again');
+        expect(JobLog.runs['again'], 2);
+        JobLog.failed.clear();
+        String id = (await Queue.driver.failed()).single.id;
+        expect(await Queue.driver.retry(id), isTrue);
+        await workUntilFailed('again');
+        // two more tries, not one: the exceptions count starts at 0
+        expect(JobLog.runs['again'], 4);
+        List<FailedJob> failed = await Queue.driver.failed();
+        expect(JobPayload.decode(failed.single.payload).exceptions, 2);
+      });
+
+      test('a retried job gets a new retryUntil window', () async {
+        await OptionsJob(key: 'window', untilMillis: 400).dispatch();
+        await workUntilFailed('window');
+        int first = JobLog.runs['window']!;
+        expect(first, greaterThan(1));
+        JobLog.failed.clear();
+        String id = (await Queue.driver.failed()).single.id;
+        // the old window is over
+        await _wait(100);
+        expect(await Queue.driver.retry(id), isTrue);
+        await workUntilFailed('window');
+        // more than one try: the old window would give only one
+        expect(JobLog.runs['window']! - first, greaterThan(1));
+      });
+
       test('without retryUntil, tries is the limit', () async {
         await OptionsJob(key: 'once').dispatch();
         await workUntilFailed('once');