Initial code for transactional outbox pattern (#13)

* Support transactional outbox pattern

* Create job ID before add call

This makes it easier to get the job ID when the actual job addition is
deferred, as when using an outbox pattern.

* finish outbox listener

* functions to add to the outbox

* notify when adding to the outbox

* cargo.toml metadata

* comment about not ready for use

* test fixes

Author: dimfeld

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "effectum",
+  "outbox",
   "stress_test"
 ]
 resolver = "2"

effectum/Cargo.toml

@@ -31,8 +31,7 @@ thiserror = "1"
 time = { version = "0.3", features = ["serde"] }
 tokio = { version = "1.32.0", features = ["rt", "macros", "time", "sync"] }
 tracing = "0.1.37"
-ulid = { version = "1.1.0", features = ["uuid"] }
-uuid = "1.4.1"
+uuid = { version = "1.7.0", features = ["v7", "serde"] }
 
 [dev-dependencies]
 color-eyre = "0.6.2"

effectum/src/add_job.rs

@@ -1,6 +1,7 @@
 use std::{borrow::Cow, time::Duration};
 
 use ahash::{HashMap, HashSet};
+use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
 use tracing::{instrument, Span};
 use uuid::Uuid;
@@ -18,8 +19,13 @@ use crate::{
 };
 
 /// A job to be submitted to the queue.
-#[derive(Debug, Clone)]
+/// Jobs are uniquely identified by their `id`, so adding a job with the same ID twice will fail.
+/// If you want to clone the same Job object and submit it multiple times, use [JobBuilder::clone_as_new]
+/// to generate a new ID with each clone.
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Job {
+    /// A unique identifier for the job.
+    pub id: Uuid,
     /// The name of the job, which matches the name used in the [JobRunner](crate::JobRunner) for the job.
     pub job_type: Cow<'static, str>,
     /// Jobs with higher `priority` will be executed first.
@@ -51,10 +57,17 @@ impl Job {
     pub fn builder(job_type: impl Into<Cow<'static, str>>) -> JobBuilder {
         JobBuilder::new(job_type)
     }
+
+    /// Clone the [Job] with a new `id`.
+    pub fn clone_as_new(&self) -> Self {
+        let mut job = self.clone();
+        job.id = Uuid::now_v7();
+        job
+    }
 }
 
 /// `Retries` controls the exponential backoff behavior when retrying failed jobs.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Retries {
     /// How many times to retry a job before it is considered to have failed permanently.
     pub max_retries: u32,
@@ -83,6 +96,7 @@ impl Default for Retries {
 impl Default for Job {
     fn default() -> Self {
         Self {
+            id: Uuid::now_v7(),
             job_type: Default::default(),
             priority: 0,
             weight: 1,
@@ -204,7 +218,7 @@ impl JobBuilder {
 /// Specified fields of a job to be updated, using the [Queue::update_job] method.
 /// All of these fields except the job ID are optional, so the update can set
 /// only the desired fields.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct JobUpdate {
     /// The ID of the job to update
     pub id: Uuid,
@@ -474,7 +488,7 @@ mod tests {
     use std::{sync::Arc, time::Duration};
 
     use temp_dir::TempDir;
-    use ulid::Ulid;
+    use uuid::Uuid;
 
     use crate::{
         test_util::{
@@ -675,7 +689,7 @@ mod tests {
         let result = test
             .queue
             .update_job(
-                JobUpdate::builder(Ulid::new().into())
+                JobUpdate::builder(Uuid::now_v7().into())
                     .run_at(test.time.now())
                     .build(),
             )
@@ -777,7 +791,7 @@ mod tests {
     #[tokio::test]
     async fn cancel_nonexistent_job() {
         let test = TestEnvironment::new().await;
-        let result = test.queue.cancel_job(Ulid::new().into()).await;
+        let result = test.queue.cancel_job(Uuid::now_v7().into()).await;
 
         assert!(matches!(result, Err(Error::NotFound)));
     }

effectum/src/db_writer/add_job.rs

@@ -47,11 +47,10 @@ pub(super) fn execute_add_job_stmt(
     now: OffsetDateTime,
     status: Option<JobState>,
 ) -> Result<(i64, Uuid)> {
-    let external_id: Uuid = ulid::Ulid::new().into();
     let run_time = job_config.run_at.unwrap_or(now).unix_timestamp();
 
     jobs_stmt.execute(named_params! {
-        "$external_id": &external_id,
+        "$external_id": &job_config.id,
         "$job_type": job_config.job_type,
         "$priority": job_config.priority,
         "$weight": job_config.weight,
@@ -70,7 +69,7 @@ pub(super) fn execute_add_job_stmt(
 
     let job_id = tx.last_insert_rowid();
 
-    Ok((job_id, external_id))
+    Ok((job_id, job_config.id))
 }
 
 pub(super) fn execute_add_active_job_stmt(

effectum/src/db_writer/recurring.rs

@@ -4,6 +4,7 @@ use rusqlite::{params, Connection, OptionalExtension, Statement};
 use time::OffsetDateTime;
 use tokio::sync::oneshot;
 use tracing::{event, Level};
+use uuid::Uuid;
 
 use super::{
     add_job::{execute_add_active_job_stmt, INSERT_ACTIVE_JOBS_QUERY},
@@ -168,9 +169,10 @@ pub(super) fn schedule_next_recurring_job(
     tx: &Connection,
     now: OffsetDateTime,
     insert_job_stmt: &mut Statement,
-    job: Job,
+    mut job: Job,
 ) -> Result<(), Error> {
     // Finally, add the version of the job that will actually run the first time.
+    job.id = Uuid::now_v7();
     let (job_id, _) = execute_add_job_stmt(tx, insert_job_stmt, &job, now, None)?;
     let mut active_insert_stmt = tx.prepare_cached(INSERT_ACTIVE_JOBS_QUERY)?;
     execute_add_active_job_stmt(&mut active_insert_stmt, job_id, &job, now)?;

effectum/src/error.rs

@@ -72,6 +72,13 @@ pub enum Error {
     RecurringJobAlreadyExists(String),
 }
 
+impl Error {
+    /// Returns if the error indicates that an update or cancel request occurred too late
+    pub fn is_update_too_late(&self) -> bool {
+        matches!(self, Error::JobRunning | Error::JobFinished)
+    }
+}
+
 impl From<InteractError> for Error {
     fn from(e: InteractError) -> Self {
         Error::DbInteract(e.to_string())

effectum/src/recurring.rs

@@ -1311,11 +1311,21 @@ mod tests {
             interval: Duration::from_secs(10),
         };
         test.queue
-            .add_recurring_job("job_id_3".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_3".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
         test.queue
-            .add_recurring_job("job_id_1".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_1".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
         test.queue
@@ -1328,7 +1338,12 @@ mod tests {
             .await
             .expect("add_recurring_job");
         test.queue
-            .add_recurring_job("job_id_2".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_2".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
 
@@ -1354,11 +1369,21 @@ mod tests {
             interval: Duration::from_secs(10),
         };
         test.queue
-            .add_recurring_job("job_id_3".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_3".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
         test.queue
-            .add_recurring_job("job_id_1".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_1".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
         test.queue
@@ -1371,7 +1396,12 @@ mod tests {
             .await
             .expect("add_recurring_job");
         test.queue
-            .add_recurring_job("job_id_2".to_string(), schedule.clone(), job.clone(), false)
+            .add_recurring_job(
+                "job_id_2".to_string(),
+                schedule.clone(),
+                job.clone_as_new(),
+                false,
+            )
             .await
             .expect("add_recurring_job");
 

effectum/src/worker.rs

@@ -483,7 +483,7 @@ mod tests {
     #[should_panic]
     async fn worker_without_jobs_should_panic() {
         let test = TestEnvironment::new().await;
-        Worker::builder(&test.queue, test.context.clone())
+        let _ = Worker::builder(&test.queue, test.context.clone())
             .build()
             .await
             .unwrap();

outbox/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "effectum-outbox"
+version = "0.1.0"
+edition = "2021"
+publish = false
+authors = ["Daniel Imfeld"]
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/dimfeld/effectum"
+keywords = ["job", "task", "queue", "sqlite"]
+categories = ["asynchronous"]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+effectum = { path = "../effectum" }
+futures = "0.3.30"
+serde = { version = "1.0.197", features = ["derive"] }
+sqlx = { version = "0.7.3", features = ["uuid", "time", "json"] }
+thiserror = "1.0.57"
+time = "0.3.34"
+tokio = { version = "1.36.0", features = ["time", "sync"] }
+tracing = { version = "0.1.40", optional = true }
+uuid = { version = "1.7.0", features = ["serde"] }
+
+[features]
+default=["postgres", "tracing"]
+postgres = ["sqlx/postgres"]
+tracing = ["dep:tracing"]

outbox/src/README.md

@@ -0,0 +1,5 @@
+# effectum-outbox
+
+An implementation of the transactional outbox pattern for Effectum. This code is not ready for use yet,
+and development is mostly on hold until I build the server mode for Effectum, at which time it will
+make more sense to finish this.

outbox/src/lib.rs

@@ -0,0 +1,10 @@
+#![warn(missing_docs)]
+
+//! Outbox pattern implementation for Effectum
+//! This is not ready for use yet.
+
+#[cfg(feature = "postgres")]
+pub mod postgres;
+
+#[cfg(feature = "postgres")]
+pub use postgres::*;