From ba5cd79190b39846c259f24e7ae787da8276f8bc Mon Sep 17 00:00:00 2001 From: mydoxie <351276057+mydoxie@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:37:03 +0700 Subject: [PATCH] Reset retry counters on queue:retry A retried job kept its old exceptions count and its old retryUntil. So it failed at its first error, or got only one try. Now queue:retry resets the tries and the exceptions count. It also reads retryUntil again from the job class, like a new dispatch. --- CHANGELOG.md | 2 +- lib/src/drivers/database_queue_driver.dart | 2 +- lib/src/drivers/redis_queue_driver.dart | 2 +- lib/src/queue.dart | 15 +++++++++++ lib/src/queue_driver.dart | 15 +++++++++++ test/job_options_test.dart | 29 ++++++++++++++++++++++ 6 files changed, 62 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b30464a..39e35ec 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/lib/src/drivers/database_queue_driver.dart b/lib/src/drivers/database_queue_driver.dart index 71c39f3..a8f5117 100644 --- a/lib/src/drivers/database_queue_driver.dart +++ b/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; }); diff --git a/lib/src/drivers/redis_queue_driver.dart b/lib/src/drivers/redis_queue_driver.dart index 2403770..12e86c4 100644 --- a/lib/src/drivers/redis_queue_driver.dart +++ b/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; } diff --git a/lib/src/queue.dart b/lib/src/queue.dart index 277d5bf..5010ead 100644 --- a/lib/src/queue.dart +++ b/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); + 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(); diff --git a/lib/src/queue_driver.dart b/lib/src/queue_driver.dart index 1fcf7c2..7a1b014 100644 --- a/lib/src/queue_driver.dart +++ b/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( diff --git a/test/job_options_test.dart b/test/job_options_test.dart index 9959437..e8134a6 100644 --- a/test/job_options_test.dart +++ b/test/job_options_test.dart @@ -233,6 +233,35 @@ void main() { expect(JobLog.failed, ['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 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');