Job options, worker limits, queue commands, fake fixes

CHANGELOG.md

@@ -1,5 +1,15 @@
 ## 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: `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.
+- New commands of `bin/queue.dart`: `clear [queue]` deletes the jobs of a queue. `restart` makes the running workers stop after their current jobs (it writes a time to the app cache, so the workers must share one cache store). `prune-failed --hours=N` deletes failed jobs older than N hours (default 24).
+- New `QueueDriver.clear()` and `QueueDriver.pruneFailed()`. The Redis and database drivers have them. A custom driver that does not override them throws an `UnsupportedError`.
+- `QueueDriver.release()` now also saves `job.payload`, so the worker can keep the exception count. A custom driver should save it too.
+- Fix: the `QueueFake` messages said "a UnusedJob". Now only "Uni" and "U" + one consonant (not "n") + vowel give "a", so it says "an UnusedJob" and still "a UniJob".
+- Fix: when `Mail.fake()` was on before `Queue.fake()`, `Queue.restore()` lost a queue set with `Mail.useQueue()`. Now `Queue.fake()` does not touch the mail queue while `Mail.fake()` is on.
+
 - Fix: the `QueueFake` check messages had bad grammar, like "a AllTypesJob" and "1 times". Now they say "an AllTypesJob" and "1 time". A name that starts with a "you" sound (like `UserJob` or `EuJob`) gets "a".
 - Fix: `Queue.restore()` did not undo the `Mail.useQueue()` call of `Queue.fake()`. With no `Queue.configure()`, `Mail.queue()` still found a queue after the test. Now `restore()` puts back the queue it had before `Queue.fake()` (also one from `Mail.useQueue()`). If it had none: no queue, or this queue when `Queue.configure()` was called.
 - New: `Queue.fake()` for tests. It keeps every dispatched job in a `QueueFake` and does not run or save them. Check them with `assertPushed<T>([test])`, `assertPushedTimes<T>(n)`, `assertNotPushed<T>()` and `assertNothingPushed()`. Read them with `pushed` (each has the job, the queue name and the delay) or `jobsOf<T>()`. It also catches `Mail.queue()` and `Mail.later()`. `dispatchSync()` still runs the job. `Queue.restore()` stops the fake.

README.md

@@ -186,6 +186,9 @@ All the settings are optional:
 | `tries` | the worker `--tries` (1) | how many times to run the job before it fails |
 | `backoff` | none (retry at once) | the wait before each retry. The last entry is used for the rest |
 | `timeout` | the worker `--timeout` (60 seconds) | the max run time of one try |
+| `retryUntil` | none | retry until this time, not `tries` times. Read once, when the job is dispatched |
+| `maxExceptions` | none | fail after this many tries that threw an error. A timeout does not count |
+| `failOnTimeout` | `false` | fail at once when a try times out, with no retry |
 | `queue` | `default` | the queue to put the job on |
 | `failed()` | does nothing | runs once, after the last try failed |
 
@@ -263,8 +266,27 @@ await SendWelcomeEmail(userId: 1).dispatch(
 );
 ```
 
+Wait until a time:
+
+```dart
+await SendWelcomeEmail(userId: 1).dispatch(
+  delayUntil: DateTime.utc(2030, 1, 1, 9),
+);
+```
+
+Give `delay` or `delayUntil`, not both.
+
 `dispatch()` returns the job id.
 
+Dispatch only when a condition is true, or false:
+
+```dart
+await SendWelcomeEmail(userId: 1).dispatchIf(user.wantsEmail);
+await SendWelcomeEmail(userId: 1).dispatchUnless(user.isBlocked);
+```
+
+They return null when the job was not dispatched.
+
 Run a job now, in this process, with no queue:
 
 ```dart
@@ -303,6 +325,9 @@ Commands:
 | `retry all` | put all failed jobs back |
 | `forget <id>` | delete one failed job |
 | `flush` | delete all failed jobs |
+| `clear [queue]` | delete the jobs of a queue. Default: `default` |
+| `restart` | stop the running workers after their current jobs |
+| `prune-failed` | delete failed jobs older than `--hours=N` (default 24) |
 
 Flags of `work`:
 
@@ -316,6 +341,12 @@ Flags of `work`:
 | `--sleep=3` | `3` | seconds to wait when the queues are empty |
 | `--stop-when-empty` | off | stop when the queues are empty |
 | `--timeout-grace=30` | off | after a timeout, wait up to this many seconds for the old run to end before the next try |
+| `--backoff=1,5,10` | none | seconds to wait before each retry of a job that does not set `backoff` |
+| `--max-jobs=1000` | off | stop after this many jobs |
+| `--max-time=3600` | off | stop after this many seconds |
+| `--memory=256` | off | stop when the process uses more than this many MB |
+
+With `--max-jobs`, `--max-time` and `--memory`, the worker stops after its running jobs. Let your process manager start it again.
 
 ```bash
 dart run bin/queue.dart work --queue=emails,default --concurrency=4
@@ -333,6 +364,12 @@ A second signal stops it at once.
 > **Restart the worker after a deploy**
 > The worker keeps your code in memory. Restart it after each deploy, or it runs the old code.
 
+`dart run bin/queue.dart restart` tells every running worker to stop after its current jobs.
+Your process manager then starts it with the new code.
+It writes a time to the app cache. Each worker reads it every 3 seconds.
+So the workers and the command must use the same cache store, eg. `file` on one machine, or `redis`.
+The `memory` store does not work for it.
+
 ### Keep it running
 
 Run the worker as its own process, next to the web server. For example with Docker Compose:

lib/src/drivers/database_queue_driver.dart

@@ -113,10 +113,11 @@ class DatabaseQueueDriver extends QueueDriver {
   @override
   Future<void> release(ReservedJob job, Duration delay) async {
     await _db.execute(
-      'UPDATE $jobsTable SET reserved_at = NULL, available_at = @at '
-      'WHERE id = @id',
+      'UPDATE $jobsTable SET reserved_at = NULL, available_at = @at, '
+      'payload = @payload WHERE id = @id',
       substitutionValues: <String, dynamic>{
         'at': QueueDriver.now() + delay.inMilliseconds,
+        'payload': job.payload.encode(),
         'id': job.ref,
       },
     );
@@ -179,6 +180,37 @@ class DatabaseQueueDriver extends QueueDriver {
     await _db.execute('DELETE FROM $failedTable');
   }
 
+  @override
+  Future<int> clear(String queue) {
+    return _deleteWhere(jobsTable, 'queue = @value', queue);
+  }
+
+  @override
+  Future<int> pruneFailed(DateTime before) {
+    return _deleteWhere(
+      failedTable,
+      'failed_at < @value',
+      before.millisecondsSinceEpoch,
+    );
+  }
+
+  /// delete the rows of [table] that match [where], and count them
+  Future<int> _deleteWhere(String table, String where, Object value) {
+    DBDriver db = _db;
+    Map<String, dynamic> values = <String, dynamic>{'value': value};
+    return db.transaction(() async {
+      List<Map<String, dynamic>> rows = await db.query(
+        'SELECT COUNT(*) AS total FROM $table WHERE $where',
+        substitutionValues: values,
+      );
+      await db.execute(
+        'DELETE FROM $table WHERE $where',
+        substitutionValues: values,
+      );
+      return _int(rows.first['total']);
+    });
+  }
+
   Future<FailedJob?> _find(DBDriver db, String id) async {
     List<Map<String, dynamic>> rows = await db.query(
       'SELECT uuid, queue, payload, exception, failed_at FROM $failedTable '

lib/src/drivers/redis_queue_driver.dart

@@ -49,10 +49,11 @@ local reserved = cjson.encode(decoded)
 redis.call('ZADD', KEYS[3], ARGV[2], reserved)
 return reserved''';
 
-  /// Moves a reserved job to the delayed set.
+  /// Moves a reserved job to the delayed set, as the new text ARGV[3]
+  /// (it can have a new exception count).
   static const String _releaseScript = '''
 redis.call('ZREM', KEYS[1], ARGV[1])
-redis.call('ZADD', KEYS[2], ARGV[2], ARGV[1])
+redis.call('ZADD', KEYS[2], ARGV[2], ARGV[3])
 return 1''';
 
   String _list(String queue) => '${prefix}queue:$queue';
@@ -106,6 +107,7 @@ return 1''';
       _delayed(queue),
       job.ref,
       QueueDriver.now() + delay.inMilliseconds,
+      job.payload.encode(),
     ]);
   }
 
@@ -155,6 +157,34 @@ return 1''';
     await redis.del(_failed);
   }
 
+  /// counts the jobs of the three keys of a queue, then deletes them
+  static const String _clearScript = '''
+local total = redis.call('LLEN', KEYS[1])
+  + redis.call('ZCARD', KEYS[2])
+  + redis.call('ZCARD', KEYS[3])
+redis.call('DEL', KEYS[1], KEYS[2], KEYS[3])
+return total''';
+
+  @override
+  Future<int> clear(String queue) async {
+    Object? total = await redis.eval(_clearScript, 3, <Object>[
+      _list(queue),
+      _delayed(queue),
+      _reserved(queue),
+    ]);
+    return int.parse('$total');
+  }
+
+  @override
+  Future<int> pruneFailed(DateTime before) async {
+    List<String> ids = <String>[
+      for (FailedJob job in await failed())
+        if (job.failedAt.isBefore(before)) job.id,
+    ];
+    if (ids.isNotEmpty) await redis.hdel(_failed, ids);
+    return ids.length;
+  }
+
   /// close the connection and the pool, with no reconnect
   @override
   Future<void> close() async {

lib/src/job.dart

@@ -70,6 +70,24 @@ abstract class DoxJob {
   /// for the old run first.
   Duration? get timeout => null;
 
+  /// Opt-in: retry the job until this time, not [tries] times. It is
+  /// read once, when the job is dispatched, and saved with the job. Like
+  /// Laravel `retryUntil`. After a failed try, the job fails when this
+  /// time has passed, else it is tried again after its [backoff].
+  /// ```
+  /// @override
+  /// DateTime? get retryUntil => DateTime.now().add(Duration(minutes: 10));
+  /// ```
+  DateTime? get retryUntil => null;
+
+  /// Opt-in: fail the job after this many tries that threw an error,
+  /// even when [tries] or [retryUntil] allow more. A timeout is not
+  /// counted. Null: no limit.
+  int? get maxExceptions => null;
+
+  /// Opt-in: fail the job at once when a try times out, with no retry.
+  bool get failOnTimeout => false;
+
   /// The queue to put the job on. `dispatch(queue:)` wins over it.
   String get queue => 'default';
 
@@ -79,15 +97,60 @@ abstract class DoxJob {
 
   /// Put the job on the queue. A worker runs it later. Returns the job
   /// id. With the `sync` driver it runs at once.
+  ///
+  /// Wait with [delay], or until the time [delayUntil]. Not both.
   /// ```
   /// await SendWelcomeEmail(userId: 1).dispatch();
   /// await SendWelcomeEmail(userId: 1).dispatch(
   ///   delay: Duration(minutes: 5),
   ///   queue: 'emails',
   /// );
+  /// await SendWelcomeEmail(userId: 1).dispatch(
+  ///   delayUntil: DateTime.utc(2030, 1, 1, 9),
+  /// );
   /// ```
-  Future<String> dispatch({Duration? delay, String? queue}) {
-    return Queue.dispatch(this, delay: delay, queue: queue);
+  Future<String> dispatch({
+    Duration? delay,
+    String? queue,
+    DateTime? delayUntil,
+  }) {
+    return Queue.dispatch(
+      this,
+      delay: delay,
+      queue: queue,
+      delayUntil: delayUntil,
+    );
+  }
+
+  /// [dispatch] when [condition] is true. Returns the job id, or null
+  /// when the job was not dispatched.
+  /// ```
+  /// await SendWelcomeEmail(userId: 1).dispatchIf(user.wantsEmail);
+  /// ```
+  Future<String?> dispatchIf(
+    bool condition, {
+    Duration? delay,
+    String? queue,
+    DateTime? delayUntil,
+  }) async {
+    if (!condition) return null;
+    return dispatch(delay: delay, queue: queue, delayUntil: delayUntil);
+  }
+
+  /// [dispatch] when [condition] is false. Returns the job id, or null
+  /// when the job was not dispatched.
+  Future<String?> dispatchUnless(
+    bool condition, {
+    Duration? delay,
+    String? queue,
+    DateTime? delayUntil,
+  }) {
+    return dispatchIf(
+      !condition,
+      delay: delay,
+      queue: queue,
+      delayUntil: delayUntil,
+    );
   }
 
   /// Run the job now, in this process, with no queue.

lib/src/queue.dart

@@ -26,10 +26,17 @@ class Queue {
   static QueueFake? _fake;
   static MailQueue? _mailQueueBeforeFake;
 
+  /// true when `Mail.fake()` was on at [fake]. Then [fake] and [restore]
+  /// leave the mail queue alone (see [fake]).
+  static bool _mailFakeWasOn = false;
+
   /// Keep dispatched jobs in a [QueueFake] and run or save none of them,
   /// so a test can check what was pushed. It needs no [configure]. It
   /// also catches `Mail.queue()` and `Mail.later()` of dox_mailer.
   /// [dispatchSync] still runs the job. [restore] undoes it.
+  ///
+  /// When `Mail.fake()` is already on, mails go to that fake, so this
+  /// does not change the mail queue. A `Mail.useQueue()` queue is kept.
   /// ```
   /// QueueFake queue = Queue.fake();
   /// await signUp('ann@example.com');
@@ -40,9 +47,14 @@ class Queue {
   static QueueFake fake() {
     QueueFake fake = QueueFake();
     // keep the mail queue set before the fake (a second fake() keeps the first)
-    if (_fake == null) _mailQueueBeforeFake = _currentMailQueue();
+    if (_fake == null) {
+      _mailFakeWasOn = _mailFakeOn();
+      _mailQueueBeforeFake = _mailFakeWasOn ? null : _currentMailQueue();
+    }
     _fake = fake;
-    Mail.useQueue(const QueueMailQueue());
+    // while Mail.fake() is on, Mail.mailQueue gives that fake, so the
+    // queue of Mail.useQueue() can not be read back: do not replace it
+    if (!_mailFakeWasOn) Mail.useQueue(const QueueMailQueue());
     return fake;
   }
 
@@ -52,22 +64,26 @@ class Queue {
   /// has no queue.
   static void restore() {
     MailQueue? before = _mailQueueBeforeFake;
+    bool keep = _mailFakeWasOn;
     _fake = null;
     _mailQueueBeforeFake = null;
+    _mailFakeWasOn = false;
+    if (keep) return;
     Mail.useQueue(before ?? (_config == null ? null : const QueueMailQueue()));
   }
 
-  /// the queue of `Mail.useQueue()`, or null. While `Mail.fake()` is on,
-  /// `Mail.mailQueue` gives that fake, which is not a queue to keep.
+  /// the queue of `Mail.useQueue()`, or null when there is none
   static MailQueue? _currentMailQueue() {
     try {
-      MailQueue queue = Mail.mailQueue;
-      return queue is ArrayTransport ? null : queue;
+      return Mail.mailQueue;
     } on StateError {
       return null;
     }
   }
 
+  /// `Mail.fake()` is on: `Mail.mailQueue` gives its [ArrayTransport]
+  static bool _mailFakeOn() => _currentMailQueue() is ArrayTransport;
+
   /// Use [config] in this isolate. `QueueProvider` calls it. Call it in
   /// tests, eg. with the `sync` driver.
   ///
@@ -96,13 +112,23 @@ class Queue {
   }
 
   /// Put [job] on the queue and return its id. [queue] wins over
-  /// `job.queue`. With a [delay], workers take it after the delay.
-  /// With the `sync` driver the job runs at once (no delay).
+  /// `job.queue`. With a [delay], workers take it after the delay. With
+  /// [delayUntil], workers take it at that time (a past time: at once).
+  /// Give [delay] or [delayUntil], not both. With the `sync` driver the
+  /// job runs at once (no delay).
   static Future<String> dispatch(
     DoxJob job, {
     Duration? delay,
     String? queue,
+    DateTime? delayUntil,
   }) async {
+    if (delayUntil != null) {
+      if (delay != null) {
+        throw ArgumentError('Give delay or delayUntil, not both.');
+      }
+      Duration wait = delayUntil.difference(DateTime.now());
+      delay = wait.isNegative ? Duration.zero : wait;
+    }
     QueueFake? fake = _fake;
     if (fake != null) {
       fake.pushed.add(PushedJob(job, queue: queue ?? job.queue, delay: delay));
@@ -122,6 +148,7 @@ class Queue {
       job: codec.name,
       queue: queue ?? job.queue,
       data: data,
+      retryUntil: job.retryUntil?.millisecondsSinceEpoch,
     );
     await driver.push(payload, delay: delay ?? Duration.zero);
     return id;