From 81ad4e8a3c32164c8e946ba7eef709bfc9c36581 Mon Sep 17 00:00:00 2001 From: mydoxie <351276057+mydoxie@users.noreply.github.com> Date: Fri, 6 Nov 2026 13:19:42 +0700 Subject: [PATCH] Add job options, worker limits and queue commands; fix fake nits - Jobs: retryUntil, maxExceptions, failOnTimeout. dispatch(delayUntil:), dispatchIf, dispatchUnless. - Worker: --backoff, --max-jobs, --max-time, --memory. - Commands: clear, restart, prune-failed. - QueueFake: 'an UnusedJob'; keep Mail.useQueue() when Mail.fake() was on. --- CHANGELOG.md | 10 + README.md | 37 +++ lib/src/drivers/database_queue_driver.dart | 36 ++- lib/src/drivers/redis_queue_driver.dart | 34 ++- lib/src/job.dart | 67 ++++- lib/src/queue.dart | 43 ++- lib/src/queue_driver.dart | 47 +++- lib/src/queue_fake.dart | 9 +- lib/src/queue_worker.dart | 123 ++++++++- lib/src/worker.dart | 107 +++++++- test/dispatch_test.dart | 71 +++++ test/driver_test.dart | 38 +++ test/job_options_test.dart | 297 +++++++++++++++++++++ test/queue_fake_test.dart | 26 +- 14 files changed, 905 insertions(+), 40 deletions(-) create mode 100644 test/job_options_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 03b5d70..b30464a 100644 --- a/CHANGELOG.md +++ b/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([test])`, `assertPushedTimes(n)`, `assertNotPushed()` and `assertNothingPushed()`. Read them with `pushed` (each has the job, the queue name and the delay) or `jobsOf()`. It also catches `Mail.queue()` and `Mail.later()`. `dispatchSync()` still runs the job. `Queue.restore()` stops the fake. diff --git a/README.md b/README.md index 9cd9cfa..a4a28d4 100644 --- a/README.md +++ b/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 ` | 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: diff --git a/lib/src/drivers/database_queue_driver.dart b/lib/src/drivers/database_queue_driver.dart index 52fdbe2..71c39f3 100644 --- a/lib/src/drivers/database_queue_driver.dart +++ b/lib/src/drivers/database_queue_driver.dart @@ -113,10 +113,11 @@ class DatabaseQueueDriver extends QueueDriver { @override Future 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: { '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 clear(String queue) { + return _deleteWhere(jobsTable, 'queue = @value', queue); + } + + @override + Future pruneFailed(DateTime before) { + return _deleteWhere( + failedTable, + 'failed_at < @value', + before.millisecondsSinceEpoch, + ); + } + + /// delete the rows of [table] that match [where], and count them + Future _deleteWhere(String table, String where, Object value) { + DBDriver db = _db; + Map values = {'value': value}; + return db.transaction(() async { + List> 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 _find(DBDriver db, String id) async { List> rows = await db.query( 'SELECT uuid, queue, payload, exception, failed_at FROM $failedTable ' diff --git a/lib/src/drivers/redis_queue_driver.dart b/lib/src/drivers/redis_queue_driver.dart index 55bb41e..2403770 100644 --- a/lib/src/drivers/redis_queue_driver.dart +++ b/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 clear(String queue) async { + Object? total = await redis.eval(_clearScript, 3, [ + _list(queue), + _delayed(queue), + _reserved(queue), + ]); + return int.parse('$total'); + } + + @override + Future pruneFailed(DateTime before) async { + List ids = [ + 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 close() async { diff --git a/lib/src/job.dart b/lib/src/job.dart index 65d17d1..dbf4ea9 100644 --- a/lib/src/job.dart +++ b/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 dispatch({Duration? delay, String? queue}) { - return Queue.dispatch(this, delay: delay, queue: queue); + Future 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 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 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. diff --git a/lib/src/queue.dart b/lib/src/queue.dart index bee90cf..277d5bf 100644 --- a/lib/src/queue.dart +++ b/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 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; diff --git a/lib/src/queue_driver.dart b/lib/src/queue_driver.dart index 00b3793..1fcf7c2 100644 --- a/lib/src/queue_driver.dart +++ b/lib/src/queue_driver.dart @@ -10,6 +10,8 @@ class JobPayload { required this.queue, required this.data, this.attempts = 0, + this.retryUntil, + this.exceptions = 0, }); /// a random id, like `0f3c...` @@ -26,6 +28,13 @@ class JobPayload { /// how many times a worker took the job final int attempts; + /// `DoxJob.retryUntil` as unix time in milliseconds, read once at + /// dispatch. Null: the job has none (also for jobs saved before it). + final int? retryUntil; + + /// how many tries threw an error, for `DoxJob.maxExceptions` + final int exceptions; + /// a new random id: 16 random bytes as hex static String newId() { Random random = Random.secure(); @@ -44,9 +53,26 @@ class JobPayload { queue: queue, data: data, attempts: attempts, + retryUntil: retryUntil, + exceptions: exceptions, ); } + /// a copy with other [exceptions] + JobPayload withExceptions(int exceptions) { + return JobPayload( + id: id, + job: job, + queue: queue, + data: data, + attempts: attempts, + retryUntil: retryUntil, + exceptions: exceptions, + ); + } + + /// [retryUntil] and [exceptions] are left out when not set, so the + /// text of a job with neither is the same as before they existed String encode() { return jsonEncode({ 'id': id, @@ -54,6 +80,8 @@ class JobPayload { 'queue': queue, 'data': data, 'attempts': attempts, + 'retryUntil': ?retryUntil, + if (exceptions > 0) 'exceptions': exceptions, }); } @@ -65,6 +93,8 @@ class JobPayload { queue: map['queue'] as String, data: map['data'] as String, attempts: (map['attempts'] as num? ?? 0).toInt(), + retryUntil: (map['retryUntil'] as num?)?.toInt(), + exceptions: (map['exceptions'] as num? ?? 0).toInt(), ); } } @@ -121,7 +151,8 @@ abstract class QueueDriver { /// the job is done: remove it Future delete(ReservedJob job); - /// try the job again after [delay] + /// try the job again after [delay]. Save `job.payload` as it is now: + /// the worker can give it a new `exceptions` count. Future release(ReservedJob job, Duration delay); /// the job failed all its tries: move it to the failed jobs @@ -139,6 +170,20 @@ abstract class QueueDriver { /// delete all failed jobs Future flush(); + /// Delete every job of [queue] that is waiting, delayed or reserved. + /// Returns how many were deleted. Failed jobs stay. The built-in + /// drivers have it, a custom driver throws an [UnsupportedError]. + Future clear(String queue) { + throw UnsupportedError('$runtimeType can not clear a queue'); + } + + /// Delete the failed jobs that failed before [before]. Returns how + /// many were deleted. The built-in drivers have it, a custom driver + /// throws an [UnsupportedError]. + Future pruneFailed(DateTime before) { + throw UnsupportedError('$runtimeType can not prune failed jobs'); + } + /// close the connection Future close() async {} diff --git a/lib/src/queue_fake.dart b/lib/src/queue_fake.dart index d01b5a3..a8d0779 100644 --- a/lib/src/queue_fake.dart +++ b/lib/src/queue_fake.dart @@ -87,15 +87,16 @@ class QueueFake { return 'Pushed: $names.'; } - /// `a SendWelcomeEmail`, `an AllTypesJob`, `a UserJob`. A vowel gives - /// "an", but "U" + consonant + vowel (User, Uni) and "Eu" sound like - /// "you", so they give "a". + /// names that start with a "you" sound: "Eu", "Uni", and "U" + one + /// consonant (not "n") + vowel, like User. Unused gives "an". // ponytail: a small rule, not a full English dictionary (hour, honest) static final RegExp _youSound = RegExp( - '^(eu|u[^aeiou][aeiou])', + '^(eu|uni|u[^aeioun][aeiou])', caseSensitive: false, ); + /// `a SendWelcomeEmail`, `an AllTypesJob`, `a UserJob`, `an UnusedJob`. + /// A vowel gives "an", but a "you" sound ([_youSound]) gives "a". static String _a(String name) { bool vowel = 'AEIOUaeiou'.contains(name.isEmpty ? 'x' : name[0]); return vowel && !_youSound.hasMatch(name) ? 'an $name' : 'a $name'; diff --git a/lib/src/queue_worker.dart b/lib/src/queue_worker.dart index b5da70c..cf6bea9 100644 --- a/lib/src/queue_worker.dart +++ b/lib/src/queue_worker.dart @@ -17,11 +17,17 @@ import 'worker.dart'; /// http server. Then it runs one command: /// - `work` (default): run jobs. Flags: `--queue=a,b`, /// `--concurrency=N`, `--once`, `--tries=N`, `--timeout=SECONDS`, -/// `--sleep=SECONDS`, `--stop-when-empty`, `--timeout-grace=SECONDS`. +/// `--sleep=SECONDS`, `--stop-when-empty`, `--timeout-grace=SECONDS`, +/// `--backoff=SECONDS[,SECONDS...]`, `--max-jobs=N`, +/// `--max-time=SECONDS`, `--memory=MB`. /// - `failed`: list the failed jobs. /// - `retry `: put failed jobs back on their queue. /// - `forget `: delete one failed job. /// - `flush`: delete all failed jobs. +/// - `clear [queue]`: delete the jobs of a queue (default: `default`). +/// - `restart`: tell the running workers to stop after their current +/// jobs, so the process manager starts them with the new code. +/// - `prune-failed`: delete failed jobs older than `--hours=N` (24). /// /// SIGINT (Ctrl+C) and SIGTERM stop the worker gracefully: it takes no /// new job and waits for the running jobs. A second signal stops it at @@ -35,6 +41,15 @@ class QueueWorker { /// then does not start the in-process worker. static bool active = false; + /// The cache key of `restart`. A worker reads it when it starts, and + /// stops when the value changes. It uses the default cache store of + /// the app, so all workers must share it (eg. `file` on one machine, + /// or `redis`). The `memory` store does not work for it. + static const String restartKey = 'dox_queue:restart'; + + /// how often a running worker reads [restartKey] + static const Duration restartCheck = Duration(seconds: 3); + static const String usage = ''' Usage: dart run bin/queue.dart [flags] @@ -44,6 +59,9 @@ Commands: retry put failed jobs back on their queue forget 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: --queue=a,b the queues to work, first one first (default: default) @@ -55,7 +73,13 @@ Flags of work: --stop-when-empty stop when the queues are empty --timeout-grace=SECONDS after a timeout, wait up to this long for the old run - to end before the next try (default: no wait)'''; + to end before the next try (default: no wait) + --backoff=SECONDS[,SECONDS...] + the wait before a retry of a job with no backoff + (default: retry at once) + --max-jobs=N stop after N jobs (default: no limit) + --max-time=SECONDS stop after this time (default: no limit) + --memory=MB stop when the process uses more memory (default: no limit)'''; /// Run the command of [args]. Returns the exit code: 0 for success. Future run(List args) async { @@ -101,6 +125,22 @@ Flags of work: await Queue.driver.flush(); stdout.writeln('Deleted all failed jobs.'); return 0; + case 'clear': + String queue = command.argument.isEmpty ? 'default' : command.argument; + int count = await Queue.driver.clear(queue); + stdout.writeln('Deleted $count jobs of the $queue queue.'); + return 0; + case 'restart': + await Cache().forever(restartKey, DateTime.now().toIso8601String()); + stdout.writeln('The workers stop after their current jobs.'); + return 0; + case 'prune-failed': + DateTime before = DateTime.now().subtract( + Duration(hours: command.hours), + ); + int count = await Queue.driver.pruneFailed(before); + stdout.writeln('Deleted $count failed jobs.'); + return 0; } return 64; } @@ -130,6 +170,34 @@ Flags of work: worker.stop(); } + // `restart` changes the key: stop like on a signal + Object? restartedAt; + bool canRestart = true; + try { + restartedAt = await Cache().get(restartKey); + } catch (error) { + canRestart = false; + Logger.warn('Queue: no `restart`, the cache can not be read: $error'); + } + bool checking = false; + Future checkRestart(Timer timer) async { + if (checking) return; + checking = true; + try { + if (await Cache().get(restartKey) == restartedAt) return; + } catch (_) { + return; // a cache error is not a restart, try again next time + } finally { + checking = false; + } + Logger.info('Queue: restart, stopping after the running jobs'); + timer.cancel(); + worker.stop(); + } + + Timer? restart = canRestart + ? Timer.periodic(restartCheck, checkRestart) + : null; signals.add(ProcessSignal.sigint.watch().listen(onSignal)); // Windows can not watch SIGTERM if (!Platform.isWindows) { @@ -140,6 +208,7 @@ Flags of work: '(${Queue.config.driver}, concurrency ${options.concurrency})', ); await worker.run(); + restart?.cancel(); for (StreamSubscription s in signals) { await s.cancel(); } @@ -207,21 +276,28 @@ Flags of work: /// A parsed `bin/queue.dart` command line. class QueueCommand { - QueueCommand(this.name, this.argument, this.options); + QueueCommand(this.name, this.argument, this.options, {this.hours = 24}); - /// `work`, `failed`, `retry`, `forget` or `flush` + /// `work`, `failed`, `retry`, `forget`, `flush`, `clear`, `restart` or + /// `prune-failed` final String name; - /// the id of `retry` and `forget` + /// the id of `retry` and `forget`, the queue of `clear` final String argument; final WorkerOptions options; + /// `--hours` of `prune-failed` + final int hours; + static const List _commands = [ 'work', 'failed', 'retry', 'forget', 'flush', + 'clear', + 'restart', + 'prune-failed', ]; /// Throws a [FormatException] with a clear message for a bad command @@ -264,10 +340,31 @@ class QueueCommand { sleep: Duration(seconds: _int(flags, 'sleep') ?? 3), stopWhenEmpty: flags.containsKey('stop-when-empty'), timeoutGrace: _seconds(flags, 'timeout-grace'), + backoff: _backoff(flags['backoff']), + maxJobs: _int(flags, 'max-jobs'), + maxTime: _seconds(flags, 'max-time'), + memory: _int(flags, 'memory'), ), + hours: _int(flags, 'hours') ?? 24, ); } + /// `--backoff=1,5,10`: seconds, one entry per retry + static List _backoff(String? value) { + if (value == null) return const []; + List waits = []; + for (String part in value.split(',')) { + int? seconds = int.tryParse(part.trim()); + if (seconds == null || seconds < 0) { + throw FormatException( + '--backoff needs seconds of 0 or more, eg. 1,5,10', + ); + } + waits.add(Duration(seconds: seconds)); + } + return waits; + } + static const List _flags = [ 'queue', 'concurrency', @@ -277,6 +374,11 @@ class QueueCommand { 'sleep', 'stop-when-empty', 'timeout-grace', + 'backoff', + 'max-jobs', + 'max-time', + 'memory', + 'hours', ]; static String _trim(String value) => value.trim(); @@ -292,7 +394,16 @@ class QueueCommand { String? value = flags[key]; if (value == null) return null; int? number = int.tryParse(value); - int min = key == 'concurrency' || key == 'tries' ? 1 : 0; + int min = + const [ + 'concurrency', + 'tries', + 'max-jobs', + 'memory', + 'max-time', + ].contains(key) + ? 1 + : 0; if (number == null || number < min) { throw FormatException('--$key needs a number of $min or more'); } diff --git a/lib/src/worker.dart b/lib/src/worker.dart index ec162ee..8e3cf38 100644 --- a/lib/src/worker.dart +++ b/lib/src/worker.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:io' show ProcessInfo; import 'job.dart'; import 'job_registry.dart'; @@ -18,6 +19,10 @@ class WorkerOptions { this.sleep = const Duration(seconds: 3), this.stopWhenEmpty = false, this.timeoutGrace, + this.backoff = const [], + this.maxJobs, + this.maxTime, + this.memory, }); /// the queues to work, first one first (`--queue=high,default`) @@ -52,6 +57,22 @@ class WorkerOptions { /// another worker can take the job again while this one waits. /// Null (the default): no wait, the retry can start at once. final Duration? timeoutGrace; + + /// the wait before each retry of a job with no `backoff` of its own + /// (`--backoff=10` or `--backoff=1,5,10`, in seconds). The last entry + /// is used for the rest. Empty (the default): retry at once. + final List backoff; + + /// stop after this many jobs (`--max-jobs=1000`). Null: no limit. + final int? maxJobs; + + /// stop after this time (`--max-time=3600`). The running jobs end + /// first. Null: no limit. + final Duration? maxTime; + + /// stop after a job when the process uses more than this many MB of + /// memory (`--memory=256`). Null: no limit. + final int? memory; } /// Takes jobs from the queue and runs them. A job that throws is tried @@ -73,11 +94,23 @@ class Worker { /// completes on [stop], to end a sleep early final Completer _wake = Completer(); + /// how many jobs this worker ran, for [WorkerOptions.maxJobs] + int _done = 0; + /// Work until [stop], or until the queues are empty with - /// `stopWhenEmpty`, or after one job with `once`. + /// `stopWhenEmpty`, or after one job with `once`, or until a limit of + /// `maxJobs`, `maxTime` or `memory` is hit. Future run() async { int loops = options.once ? 1 : options.concurrency; + Duration? maxTime = options.maxTime; + Timer? timer = maxTime == null ? null : Timer(maxTime, _timeIsUp); await Future.wait(>[for (int i = 0; i < loops; i++) _loop()]); + timer?.cancel(); + } + + void _timeIsUp() { + QueueLog.info('Queue: max time ${options.maxTime} reached, stopping'); + stop(); } /// Stop taking jobs. [run] ends when the running jobs are done. @@ -107,6 +140,22 @@ class Worker { QueueLog.danger('Queue: ${job.payload.job} error: $error', '$stack'); } if (options.once) return; + _checkLimits(); + } + } + + /// stop on [WorkerOptions.maxJobs] or [WorkerOptions.memory] + void _checkLimits() { + _done++; + int? maxJobs = options.maxJobs; + if (maxJobs != null && _done >= maxJobs && !_stopping) { + QueueLog.info('Queue: ran $_done jobs (max $maxJobs), stopping'); + stop(); + } + int? memory = options.memory; + if (memory != null && ProcessInfo.currentRss > memory * 1024 * 1024) { + QueueLog.info('Queue: memory is over $memory MB, stopping'); + stop(); } } @@ -140,12 +189,16 @@ class Worker { return; } int tries = job.tries ?? options.tries; - if (reserved.attempts > tries) { + if (reserved.attempts > 1 && + _noTryLeft(job, payload, reserved.attempts - 1)) { // a worker took it before and did not finish (it crashed or the - // reservation timed out) + // reservation timed out), or its retryUntil time passed StateError error = StateError( - '$label was tried ${reserved.attempts - 1} times and did not finish. ' - 'The worker may have crashed, or the job ran longer than retryAfter.', + payload.retryUntil != null + ? '$label is past its retryUntil time.' + : '$label was tried ${reserved.attempts - 1} times and did not ' + 'finish. The worker may have crashed, or the job ran ' + 'longer than retryAfter.', ); await _fail(reserved, job, label, error, StackTrace.current); return; @@ -159,20 +212,33 @@ class Worker { await run.timeout(timeout); } catch (error, stack) { Object reason = error; - if (error is TimeoutException) { + bool timedOut = error is TimeoutException; + int exceptions = payload.exceptions; + if (timedOut) { reason = TimeoutException('$label timed out after $timeout', timeout); if (run != null) await _waitForTimedOutRun(run, label); + } else { + exceptions++; } - if (reserved.attempts >= tries) { - await _fail(reserved, job, label, reason, stack); + ReservedJob next = ReservedJob( + payload.withExceptions(exceptions), + attempts: reserved.attempts, + ref: reserved.ref, + ); + int? maxExceptions = job.maxExceptions; + if ((timedOut && job.failOnTimeout) || + (maxExceptions != null && exceptions >= maxExceptions) || + _noTryLeft(job, payload, reserved.attempts)) { + await _fail(next, job, label, reason, stack); return; } - Duration wait = backoffFor(job, reserved.attempts); + Duration wait = backoffFor(job, reserved.attempts, options.backoff); + String of = payload.retryUntil == null ? ' of $tries' : ''; QueueLog.warn( - 'Queue: $label failed try ${reserved.attempts} of $tries, ' + 'Queue: $label failed try ${reserved.attempts}$of, ' 'next try in $wait: $reason', ); - await driver.release(reserved, wait); + await driver.release(next, wait); return; } await driver.delete(reserved); @@ -200,9 +266,22 @@ class Worker { static bool _endedWithError(Object _) => true; static bool _stillRunning() => false; - /// the wait before the next try, after try number [attempts] - static Duration backoffFor(DoxJob job, int attempts) { - List backoff = job.backoff; + /// No try is left after try number [attempts]: the time of + /// `retryUntil` has passed, or else all `tries` are used. + bool _noTryLeft(DoxJob job, JobPayload payload, int attempts) { + int? until = payload.retryUntil; + if (until != null) return QueueDriver.now() >= until; + return attempts >= (job.tries ?? options.tries); + } + + /// the wait before the next try, after try number [attempts]. A job + /// with no `backoff` uses [fallback] (the worker `--backoff`). + static Duration backoffFor( + DoxJob job, + int attempts, [ + List fallback = const [], + ]) { + List backoff = job.backoff.isEmpty ? fallback : job.backoff; if (backoff.isEmpty) return Duration.zero; int index = attempts - 1; return backoff[index < backoff.length ? index : backoff.length - 1]; diff --git a/test/dispatch_test.dart b/test/dispatch_test.dart index 64db683..5f92625 100644 --- a/test/dispatch_test.dart +++ b/test/dispatch_test.dart @@ -186,6 +186,54 @@ void main() { expect(await Queue.driver.failed(), isEmpty); }); + test('clear and prune-failed', () async { + Queue.configure(config); + await SendWelcomeEmail(userId: 1).dispatch(); + await SendWelcomeEmail(userId: 2).dispatch(queue: 'emails'); + await FlakyJob( + key: 'p', + failTimes: 9, + maxTries: 1, + ).dispatch(queue: 'flaky'); + expect(await run(['work', '--queue=flaky', '--once']), 0); + expect(await run(['clear']), 0); + Queue.configure(config); + expect(await Queue.driver.pop('default'), isNull); + expect(await Queue.driver.pop('emails'), isNotNull); + expect(await run(['clear', 'emails']), 0); + Queue.configure(config); + expect(await Queue.driver.pop('emails'), isNull); + // the failed job is new: 1 hour keeps it, 0 hours deletes it + expect(await run(['prune-failed', '--hours=1']), 0); + Queue.configure(config); + expect(await Queue.driver.failed(), hasLength(1)); + expect(await run(['prune-failed', '--hours=0']), 0); + Queue.configure(config); + expect(await Queue.driver.failed(), isEmpty); + }); + + test('restart stops a running worker', () async { + // a memory cache, so the test writes no files + Dox().cache = CacheManager( + const CacheConfig( + defaultStore: 'memory', + stores: >{ + 'memory': {'driver': 'memory'}, + }, + ), + ); + Future worker = run(['work']); + await _wait(500); + // like `restart` from another process: the shared cache changes + await Cache().forever(QueueWorker.restartKey, 'restart 1'); + expect( + await worker.timeout( + QueueWorker.restartCheck + const Duration(seconds: 5), + ), + 0, + ); + }); + test('the sync driver needs no worker', () async { int code = await QueueWorker(app: _app(QueueConfig())) .run(['work']); @@ -219,6 +267,10 @@ void main() { '--sleep=1', '--stop-when-empty', '--timeout-grace=5', + '--backoff=1,5', + '--max-jobs=10', + '--max-time=60', + '--memory=128', ]); expect(command.name, 'work'); WorkerOptions options = command.options; @@ -230,7 +282,26 @@ void main() { expect(options.sleep, const Duration(seconds: 1)); expect(options.stopWhenEmpty, isTrue); expect(options.timeoutGrace, const Duration(seconds: 5)); + expect(options.backoff, const [ + Duration(seconds: 1), + Duration(seconds: 5), + ]); + expect(options.maxJobs, 10); + expect(options.maxTime, const Duration(seconds: 60)); + expect(options.memory, 128); expect(QueueCommand.parse([]).options.timeoutGrace, isNull); + expect(QueueCommand.parse([]).options.maxJobs, isNull); + expect(QueueCommand.parse(['prune-failed']).hours, 24); + expect(QueueCommand.parse(['prune-failed', '--hours=2']).hours, 2); + expect(QueueCommand.parse(['clear', 'emails']).argument, 'emails'); + expect( + () => QueueCommand.parse(['--backoff=1,x']), + throwsFormatException, + ); + expect( + () => QueueCommand.parse(['--max-jobs=0']), + throwsFormatException, + ); expect(QueueCommand.parse([]).name, 'work'); expect(QueueCommand.parse(['retry', 'all']).argument, 'all'); }); diff --git a/test/driver_test.dart b/test/driver_test.dart index 46472a3..75d4b65 100644 --- a/test/driver_test.dart +++ b/test/driver_test.dart @@ -126,6 +126,44 @@ void main() { await driver.flush(); expect(await driver.failed(), isEmpty); }); + + test('clear deletes the jobs of one queue', () async { + await driver.push(_payload('a')); + await driver.push(_payload('b'), delay: const Duration(minutes: 1)); + await driver.push(_payload('c')); + await driver.pop('default'); + await driver.push(_payload('d', queue: 'emails')); + // one ready, one delayed, one reserved + expect(await driver.clear('default'), 3); + expect(await driver.pop('default'), isNull); + expect(await driver.clear('default'), 0); + expect((await driver.pop('emails'))!.payload.id, 'd'); + }); + + test('pruneFailed deletes the old failed jobs', () async { + await driver.push(_payload('a')); + await driver.fail((await driver.pop('default'))!, 'boom'); + DateTime past = DateTime.now().subtract(const Duration(hours: 1)); + expect(await driver.pruneFailed(past), 0); + expect(await driver.failed(), hasLength(1)); + DateTime later = DateTime.now().add(const Duration(seconds: 1)); + expect(await driver.pruneFailed(later), 1); + expect(await driver.failed(), isEmpty); + }); + + test('release saves the new exception count', () async { + await driver.push(_payload('a')); + ReservedJob job = (await driver.pop('default'))!; + ReservedJob next = ReservedJob( + job.payload.withExceptions(2), + attempts: job.attempts, + ref: job.ref, + ); + await driver.release(next, Duration.zero); + ReservedJob again = (await driver.pop('default'))!; + expect(again.payload.exceptions, 2); + expect(again.attempts, 2); + }); }, skip: store.skip); } } diff --git a/test/job_options_test.dart b/test/job_options_test.dart new file mode 100644 index 0000000..9959437 --- /dev/null +++ b/test/job_options_test.dart @@ -0,0 +1,297 @@ +import 'dart:async'; + +import 'package:dox_queue/dox_queue.dart'; +import 'package:test/test.dart'; + +import 'fixtures/jobs.g.dart'; +import 'fixtures/test_jobs.dart'; +import 'support.dart'; + +/// a job with the opt-in options. It throws, or waits past its timeout. +class OptionsJob extends DoxJob { + OptionsJob({ + required this.key, + this.maxTries = 1, + this.untilMillis, + this.exceptions, + this.onTimeout = false, + this.slow = false, + }); + + final String key; + final int maxTries; + final int? untilMillis; + final int? exceptions; + final bool onTimeout; + final bool slow; + + @override + int get tries => maxTries; + + @override + DateTime? get retryUntil { + int? millis = untilMillis; + return millis == null + ? null + : DateTime.now().add(Duration(milliseconds: millis)); + } + + @override + int? get maxExceptions => exceptions; + + @override + bool get failOnTimeout => onTimeout; + + @override + Duration? get timeout => const Duration(milliseconds: 50); + + @override + List get backoff => const [Duration(milliseconds: 50)]; + + @override + Future handle() async { + JobLog.runs[key] = (JobLog.runs[key] ?? 0) + 1; + if (slow) { + await Future.delayed(const Duration(milliseconds: 150)); + return; + } + throw StateError('options $key'); + } + + @override + void failed(Object error, StackTrace stack) { + JobLog.failed.add(key); + } +} + +Map _encode(DoxJob job) { + OptionsJob value = job as OptionsJob; + return { + 'key': value.key, + 'maxTries': value.maxTries, + 'untilMillis': value.untilMillis, + 'exceptions': value.exceptions, + 'onTimeout': value.onTimeout, + 'slow': value.slow, + }; +} + +DoxJob _decode(Map data) { + return OptionsJob( + key: data['key'] as String, + maxTries: data['maxTries'] as int, + untilMillis: data['untilMillis'] as int?, + exceptions: data['exceptions'] as int?, + onTimeout: data['onTimeout'] as bool, + slow: data['slow'] as bool, + ); +} + +JobRegistry _jobs = JobRegistry([ + ...jobRegistry.codecs, + const JobCodec('OptionsJob', OptionsJob, _encode, _decode), +]); + +Future _wait(int millis) { + return Future.delayed(Duration(milliseconds: millis)); +} + +void main() { + setUpAll(setUpTestApp); + + test('a payload of an old row has no retryUntil and no exceptions', () { + JobPayload old = JobPayload.decode( + '{"id":"a","job":"J","queue":"default","data":"{}","attempts":2}', + ); + expect(old.retryUntil, isNull); + expect(old.exceptions, 0); + expect(old.attempts, 2); + // a job with neither is saved as before + expect(old.encode(), isNot(contains('retryUntil'))); + expect(old.encode(), isNot(contains('exceptions'))); + JobPayload next = JobPayload.decode( + JobPayload( + id: 'b', + job: 'J', + queue: 'q', + data: '{}', + retryUntil: 99, + ).withExceptions(3).withAttempts(1).encode(), + ); + expect(next.retryUntil, 99); + expect(next.exceptions, 3); + expect(next.attempts, 1); + }); + + test('Worker.backoffFor uses the worker backoff for a job with none', () { + SendWelcomeEmail job = SendWelcomeEmail(userId: 1); + List fallback = const [ + Duration(seconds: 1), + Duration(seconds: 5), + ]; + expect(Worker.backoffFor(job, 1), Duration.zero); + expect(Worker.backoffFor(job, 1, fallback), Duration(seconds: 1)); + expect(Worker.backoffFor(job, 9, fallback), Duration(seconds: 5)); + FlakyJob flaky = FlakyJob(key: 'a', failTimes: 1); + expect(Worker.backoffFor(flaky, 1, fallback), flaky.backoff.first); + }); + + group('fake', () { + late QueueFake queue; + + setUp(() => queue = Queue.fake()); + + tearDown(Queue.restore); + + test('dispatchIf and dispatchUnless', () async { + expect(await SendWelcomeEmail(userId: 1).dispatchIf(false), isNull); + expect(await SendWelcomeEmail(userId: 2).dispatchUnless(true), isNull); + queue.assertNothingPushed(); + expect(await SendWelcomeEmail(userId: 3).dispatchIf(true), isNotNull); + expect( + await SendWelcomeEmail(userId: 4) + .dispatchUnless(false, queue: 'emails'), + isNotNull, + ); + expect(queue.jobsOf().map(_userId), [3, 4]); + expect(queue.pushed.last.queue, 'emails'); + }); + + test('delayUntil turns into a delay', () async { + await SendWelcomeEmail(userId: 1) + .dispatch(delayUntil: DateTime.now().add(Duration(minutes: 10))); + Duration delay = queue.pushed.single.delay!; + expect(delay, greaterThan(Duration(minutes: 9))); + expect(delay, lessThanOrEqualTo(Duration(minutes: 10))); + // a past time: at once + await SendWelcomeEmail(userId: 2) + .dispatch(delayUntil: DateTime.now().subtract(Duration(minutes: 1))); + expect(queue.pushed.last.delay, Duration.zero); + expect( + () => + SendWelcomeEmail(userId: 3) + .dispatch(delay: Duration.zero, delayUntil: DateTime.now()), + throwsArgumentError, + ); + }); + }); + + for (TestStore store in testStores) { + group('${store.name} job options', () { + setUp(() async { + await store.reset(); + JobLog.clear(); + Queue.configure( + store.config( + jobs: _jobs, + retryAfter: const Duration(milliseconds: 500), + ), + ); + }); + + tearDown(Queue.close); + + Worker worker([WorkerOptions? options]) { + return Worker( + driver: Queue.driver, + jobs: _jobs, + options: options ?? const WorkerOptions(stopWhenEmpty: true), + ); + } + + /// work until [key] failed, or 5 seconds + Future workUntilFailed(String key) async { + Stopwatch watch = Stopwatch()..start(); + while (!JobLog.failed.contains(key) && watch.elapsed.inSeconds < 5) { + await worker().run(); + await _wait(20); + } + } + + test('delayUntil waits until the time', () async { + await SendWelcomeEmail(userId: 1).dispatch( + delayUntil: DateTime.now().add(const Duration(milliseconds: 300)), + ); + expect(await Queue.driver.pop('default'), isNull); + await _wait(400); + expect(await Queue.driver.pop('default'), isNotNull); + }); + + test('maxExceptions fails before the tries are used', () async { + await OptionsJob(key: 'max', maxTries: 10, exceptions: 2).dispatch(); + await workUntilFailed('max'); + expect(JobLog.runs['max'], 2); + List failed = await Queue.driver.failed(); + expect(JobPayload.decode(failed.single.payload).exceptions, 2); + }); + + test('retryUntil retries past tries until the time', () async { + await OptionsJob(key: 'until', untilMillis: 400).dispatch(); + await workUntilFailed('until'); + // tries is 1, but it ran until the time passed + expect(JobLog.runs['until'], greaterThan(1)); + expect(JobLog.failed, ['until']); + }); + + test('without retryUntil, tries is the limit', () async { + await OptionsJob(key: 'once').dispatch(); + await workUntilFailed('once'); + expect(JobLog.runs['once'], 1); + }); + + test('failOnTimeout fails at once on a timeout', () async { + await OptionsJob( + key: 'slow', + maxTries: 3, + onTimeout: true, + slow: true, + ).dispatch(); + await workUntilFailed('slow'); + expect(JobLog.runs['slow'], 1); + List failed = await Queue.driver.failed(); + expect(failed.single.error, contains('timed out')); + }); + + test('a timeout is not an exception for maxExceptions', () async { + await OptionsJob( + key: 'slow2', + maxTries: 2, + exceptions: 1, + slow: true, + ).dispatch(); + await workUntilFailed('slow2'); + // maxExceptions 1 did not stop it: both tries ran + expect(JobLog.runs['slow2'], 2); + }); + + test('--max-jobs stops the worker', () async { + for (int i = 1; i <= 3; i++) { + await SendWelcomeEmail(userId: i).dispatch(); + } + await worker(const WorkerOptions(maxJobs: 2)).run(); + expect(JobLog.handled, ['welcome 1', 'welcome 2']); + expect(await Queue.driver.pop('default'), isNotNull); + }); + + test('--max-time stops an idle worker', () async { + Stopwatch watch = Stopwatch()..start(); + await worker(const WorkerOptions(maxTime: Duration(milliseconds: 200))) + .run() + .timeout(const Duration(seconds: 5)); + expect(watch.elapsedMilliseconds, lessThan(2000)); + }); + + test('--memory stops the worker after a job', () async { + await SendWelcomeEmail(userId: 1).dispatch(); + await SendWelcomeEmail(userId: 2).dispatch(); + // 1 MB: every process uses more + await worker(const WorkerOptions(memory: 1)) + .run() + .timeout(const Duration(seconds: 5)); + expect(JobLog.handled, ['welcome 1']); + }); + }, skip: store.skip); + } +} + +int _userId(SendWelcomeEmail job) => job.userId; diff --git a/test/queue_fake_test.dart b/test/queue_fake_test.dart index 95db25b..b24814c 100644 --- a/test/queue_fake_test.dart +++ b/test/queue_fake_test.dart @@ -34,6 +34,11 @@ class UploadJob extends DoxJob { Future handle() async {} } +class UnusedJob extends DoxJob { + @override + Future handle() async {} +} + class OrderJob extends DoxJob { @override Future handle() async {} @@ -50,6 +55,8 @@ class _CustomQueue implements MailQueue { }) async => 'id'; } +void _noMailQueue() => Mail.useQueue(null); + bool _isUser2(SendWelcomeEmail job) => job.userId == 2; bool _isUser9(SendWelcomeEmail job) => job.userId == 9; @@ -191,6 +198,10 @@ void main() { _failure(queue.assertPushed), startsWith('Expected an UploadJob job'), ); + expect( + _failure(queue.assertPushed), + startsWith('Expected an UnusedJob job'), + ); expect( _failure(queue.assertPushed), startsWith('Expected an OrderJob job'), @@ -201,10 +212,23 @@ void main() { Queue.restore(); _CustomQueue custom = _CustomQueue(); Mail.useQueue(custom); + addTearDown(_noMailQueue); Queue.fake(); expect(Mail.mailQueue, isNot(same(custom))); Queue.restore(); expect(Mail.mailQueue, same(custom)); - Mail.useQueue(null); + }); + + test('keeps a Mail.useQueue() when Mail.fake() was on first', () { + Queue.restore(); + _CustomQueue custom = _CustomQueue(); + Mail.useQueue(custom); + addTearDown(_noMailQueue); + Mail.fake(); + addTearDown(Mail.restore); + Queue.fake(); + Queue.restore(); + Mail.restore(); + expect(Mail.mailQueue, same(custom)); }); }