Adding a 'run after elapsed' feature for enqueuing

[isaacdonaldson]

Something I use quite often with other queue implementations is the ability to specify when to run a workload later that is not periodic.

What are your thoughts on adding support for a enqueue_at(Job{...}, std::time::Duration::from_secs(600)) where you can how long to wait before running the job?

[maxcountryman]

Sure. Would this be different from delay?

[isaacdonaldson]

Ah I didn't see that while going through the docs, apologies.

I think the difference I discern is that delay is on the Task trait, so inherently bound to the definition of what the task is, while enqueue_after(..) (or equivalent named operation) is tied to each task object/instantiation. The difference being the ability to specify enqueue_after(WelcomeTask{}, Duration::from_secs(5)), but also be able to specify enqueue_after(WelcomeTask{}, Duration::from_secs(5_000)) without having to define a WelcomeSoonJob and a WelcomeNotVerySoonJob.

Please correct me if my understanding is wrong.

[maxcountryman]

I think that distinction does make sense.

Thinking about how this would be implemented, it's unclear to me if this requires a new column on the task table (e.g. available_at) or if it's sufficient to add the duration given to enqueue_after to the task's delay duration. (Earlier iterations did have an available_at column, but I ended up removing it in favor of using the delay interval with created_at.)

[maxcountryman]

Here's one potential implementation:

diff --git a/src/queue.rs b/src/queue.rs
index 84051e7..64c14d8 100644
--- a/src/queue.rs
+++ b/src/queue.rs
@@ -326,6 +326,37 @@ impl<T: Task> Queue<T> {
     /// [ULID]: https://github.com/ulid/spec?tab=readme-ov-file#specification
     #[instrument(skip(self, executor, task, input), fields(task.id = tracing::field::Empty), err)]
     pub async fn enqueue<'a, E>(&self, executor: E, task: &T, input: T::Input) -> Result<TaskId>
+    where
+        E: PgExecutor<'a>,
+    {
+        self.enqueue_with_delay(executor, task, input, task.delay())
+            .await
+    }
+
+    pub async fn enqueue_after<'a, E>(
+        &self,
+        executor: E,
+        task: &T,
+        input: T::Input,
+        delay: Span,
+    ) -> Result<TaskId>
+    where
+        E: PgExecutor<'a>,
+    {
+        let calculated_delay = task.delay().checked_add(delay)?;
+        self.enqueue_with_delay(executor, task, input, calculated_delay)
+            .await
+    }
+
+    // Explicitly provide for a delay so that we can also facilitate calculated
+    // retries, i.e. `enqueue_after`.
+    async fn enqueue_with_delay<'a, E>(
+        &self,
+        executor: E,
+        task: &T,
+        input: T::Input,
+        delay: Span,
+    ) -> Result<TaskId>
     where
         E: PgExecutor<'a>,
     {
@@ -336,7 +367,6 @@ impl<T: Task> Queue<T> {
         let retry_policy = task.retry_policy();
         let timeout = task.timeout();
         let ttl = task.ttl();
-        let delay = task.delay();
         let concurrency_key = task.concurrency_key();
         let priority = task.priority();

[isaacdonaldson]

Nice! I like how simple a change that is, super easy to make sense of. I agree, it makes sense to add the passed in delay to the delay from the task definition, as well. (although a note about that in the docs could be nice 😄 )

[maxcountryman]

I was thinking the same re docs, since it is a bit nuanced if you have both a delay and then use this method.

I'll go ahead and add this and it should be incorporated into the next release.