---
# System prepended metadata

title: Job Queue

---

# Job Queue

## Changelog 

* 2024-08-12: Add [the "Better" section](#Better) to demonstrate general and "`JSONB` column approach"-specific improvements you can possibly make.
* 2024-08-06: Initial version

<details>
    
<summary>Code from tweets</summary>
    
<img src="https://hackmd.io/_uploads/HkCG851qR.png" />
<img src="https://hackmd.io/_uploads/r1yr8qycA.jpg" />
    
</details>

A few observations first:

1. `enum Job` has two variants with the same contents:   
   * `ProcessMetricsUpload(JobDetails)`
   * `SendVerificationEmail(JobDetails)`
  
   Sometimes it totally makes sense to do that, other times you'll do just fine with a simple
   ```rust=
   struct Job {
       kind: JobKind,
   }

   enum JobKind {
       ProcessMetricsUpload,
       SendVerificationEmail,
   }
   ```
  
   NB: This "kind-based" approach won't work if you want to store different data depending on the kind:
   ```rust=
   enum Job {
       ProcessMetricsUpload { metric_name: String },
       SendVerificationEmail { email_address: EmailAddress },
   }
   ```
   
   Based on the code I've seen, it feels like `kind` would work well enough, but I won't make any assumptions to cover all possible use cases.
   
2. `struct Database` doesn't do much, it mostly copies the interface of `sqlx`. So one could say it's a level of inderection without a useful abstraction to leverage: even `query_one` and `query` methods are dependent on `sqlx` and `Postgres` functionality.

   If no abstraction is needed (i.e. you're not trying to follow "clean architecture" rules), you can just pass and `.clone()` `sqlx::PgPool` everywhere:
   ```rust=
   struct Queue {
       pg_pool: sqlx::PgPool,
       // Other fields...
   }
   ```
   

Now, based on observations above, we have three options available to us if we want to use compile-time type checking via `sqlx::query!` and `sqlx::query_as!` macros.

## 1. Kind-based approach

```rust=
#[derive(sqlx::Type)]
#[sqlx(type_name = "job_type", rename_all = "snake_case")]
pub enum JobType {
    ProcessMetricsUpload,
    SendVerificationEmail,
}

// Notice how we don't need to derive anything here
// (or implement any trait, for that matter).
struct Job {
    id: String,
    r#type: JobType,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
}

async fn get_available_jobs(&self) -> Result<Vec<Job>> {
    sqlx::query_as!(
        Job,
        r#"
            SELECT id
                 , type AS "r#type: JobType"
                 , locked_at
                 , created_at
                 , updated_at
            FROM job_queue
        "#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)
}
```

The `AS "r#type: JobType"` annotation is important! It helps `sqlx` to map `job_type` (a type stored directly in Postgres) to our `enum JobType`. Why do need need to explicitly repeat `JobType` here? `sqlx` doesn't assume there is only one possible way to convert Postgres's `job_type` into some arbitrary Rust type (there could be many!), so we nudge it a little.

## 2. Enum-based approach

More type safety, nice! We will need an intermediate struct to store raw Postgres data, which we can easily convert into our "domain-level" (sorry) struct.

```rust=
////////////////////////////////////////////////////////////////////////////////
// `job.rs`

enum Job {
    ProcessMetricsUpload(JobDetails),
    SendVerificationEmail(JobDetails),
}

struct JobDetails {
    id: String,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
}

////////////////////////////////////////////////////////////////////////////////
// `queue.rs`

async fn get_available_jobs(&self) -> Result<Vec<Job>> {
    let jobs = sqlx::query_as!(
        QueryJob,
        r#"
            SELECT id
                 , type AS "r#type: QueryJobType"
                 , locked_at
                 , created_at
                 , updated_at
            FROM job_queue 
        "#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)?;
    
    Ok(jobs
        .into_iter()
        .map(Job::from) // Implementation below.
        .collect())
}

// I usually name such struct with `Query` prefix
// to signal that this struct is used only for
// `SELECT`-ing data from Postgres and nothing else.
struct QueryJob {
    id: String,
    r#type: QueryJobType,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
}

#[derive(sqlx::Type)]
#[sqlx(type_name = "job_type", rename_all = "snake_case")]
enum QueryJobType {
    ProcessMetricsUpload,
    SendVerificationEmail,
}

// With this, we can write `query_job.into()` to get a `Job`.
// Or, alternatively, `Job::from(query_job)`.
impl From<QueryJob> for Job {
    fn from(j: QueryJob) -> Job {
        let details = JobDetails {
            id: j.id,
            locked_at: j.locked_at,
            created_at: j.created_at,
            updated_at: j.updated_at,
        };
        
        match j.r#type {
            QueryJobType::ProcessMetricsUpload => Job::ProcessMetricsUpload(details),
            QueryJobType::SendVerificationEmail => Job::SendVerificationEmail(details),
        }
    }
}
```

## 3. Advanced enums

I think it's fair to assume some day different types of jobs will need different data. Without coming up with some crazy schemas (e.g. using a table per type), we can store all the data in a single table. On the spot, I can think of two ways:

### Many columns

Store common data in standalone columns
```psql=
  id VARCHAR(64) NOT NULL
, locked_at TIMESTAMPTZ
, created_at TIMESTAMPTZ
, updated_at TIMESTAMPTZ
```
   
and store type-related data in optional columns for each type:
```psql=
-- Metrics upload.
, metrics_upload__metric_name TEXT
, metrics_upload__uploaded_at TIMESTAMPTZ

-- Verification emails.
, verification_email__email_address TEXT
, verification_email__sent_at       TIMESTAMPTZ
```
   
Now you can declare `QueryJob` with all the columns:
```rust=
////////////////////////////////////////////////////////////////////////////////
// `job.rs`

// Let's say some type-specific fields are required 
// (e.g. `metric_name` or `email_address`).
enum Job {
    ProcessMetricsUpload {
        details: JobDetails,
        metric_name: String,
        uploaded_at: Option<DateTime<Utc>>,
    },
    
    SendVerificationEmail {
        details: JobDetails,
        email_address: String,
        sent_at: Option<DateTime<Utc>>,
    },
}

struct JobDetails {
    id: String,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
}

////////////////////////////////////////////////////////////////////////////////
// `queue.rs`

struct QueryJob {
    id: String,
    r#type: QueryJobType,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
    
    // Metrics upload.
    metric_name: Option<String>, // Optional due to Postgres column being `NULL`-able.
    metric_uploaded_at: Option<DateTime<Utc>>,
    
    // Verification emails.
    verification_email_address: Option<String>, // Optional due to Postgres column being `NULL`-able.
    verification_email_sent_at: Option<DateTime<Utc>>,
    
}

#[derive(sqlx::Type)]
#[sqlx(type_name = "job_type", rename_all = "snake_case")]
enum QueryJobType {
    ProcessMetricsUpload,
    SendVerificationEmail,
}

async fn get_available_jobs(&self) -> Result<Vec<Job>> {
    let jobs = sqlx::query_as!(
        QueryJob,
        r#"
            SELECT id
                 , type AS "r#type: QueryJobType"
                 , locked_at
                 , created_at
                 , updated_at
        
                 , metrics_upload__metric_name AS metric_name
                 , metrics_upload__uploaded_at AS metric_uploaded_at
        
                 , verification_email__email_address AS verification_email_address
                 , verification_email__sent_at       AS verification_email_sent_at
            FROM job_queue 
        "#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)?;
    
    let jobs = jobs
        .into_iter()
        .map(Job::try_from)
        // See: https://doc.rust-lang.org/rust-by-example/error/iter_result.html#fail-the-entire-operation-with-collect
        .collect::<Result<Vec<Job>>()?;
    
    Ok(jobs)
}

impl TryFrom<QueryJob> for Job {
    type Error = crate::Error;
    
    fn try_from(j: QueryJob) -> Result<Job, crate::Error> {
        let details = JobDetails {
            id: j.id,
            locked_at: j.locked_at,
            created_at: j.created_at,
            updated_at: j.updated_at,
        };
        
        match j.r#type {
            QueryJobType::ProcessMetricsUpload => {
                // Manually process every required field here.
                // 
                // `Error::JobIntergrity` is a new error variant
                // to help you debug errors when they occur.
                // Be specific so that future you won't spend
                // more than one or two minutes fixing possible issue.
                let metric_name = j
                    .metric_name
                    .ok_or(crate::Error::JobIntegrity {
                        missing_field: "metric_name",
                    })?;
                
                Ok(Job::ProcessMetricsUpload {
                    details,
                    metric_name,
                    metric_uploaded_at: j.metric_uploaded_at,
                })
            }
            QueryJobType::SendVerificationEmail => {
                let verification_email_address = j
                    .verification_email_address
                    .ok_or(crate::Error::JobIntegrity {
                        missing_field: "verification_email_address",
                    })?
                
                Ok(Job::SendVerificationEmail {
                    details,
                    verification_email_address,
                    verification_email_sent_at: j.verification_email_sent_at,
                })
            }
        }
    }
}
```

### Single JSON(B) column

Store common data in standalone columns
```psql=
  id VARCHAR(64) NOT NULL
, locked_at TIMESTAMPTZ
, created_at TIMESTAMPTZ
, updated_at TIMESTAMPTZ
```
   
and store type-related data in a single JSONB column:
```psql=
, data JSONB (NOT NULL?)
```

```rust=
////////////////////////////////////////////////////////////////////////////////
// `job.rs`

enum Job {
    ProcessMetricsUpload {
        details: JobDetails,
        metric_name: String,
        uploaded_at: Option<DateTime<Utc>>,
    },
    
    SendVerificationEmail {
        details: JobDetails,
        email_address: String,
        sent_at: Option<DateTime<Utc>>,
    },
}

struct JobDetails {
    id: String,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
}

////////////////////////////////////////////////////////////////////////////////
// `queue.rs`

struct QueryJob {
    id: String,
    r#type: QueryJobType,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
    
    // Type-specific data.
    data: serde_json::Value,
}

#[derive(sqlx::Type)]
#[sqlx(type_name = "job_type", rename_all = "snake_case")]
enum QueryJobType {
    ProcessMetricsUpload,
    SendVerificationEmail,
}

async fn get_available_jobs(&self) -> Result<Vec<Job>> {
    let jobs = sqlx::query_as!(
        QueryJob,
        r#"
            SELECT id
                 , type AS "r#type: QueryJobType"
                 , locked_at
                 , created_at
                 , updated_at

                 , data
            FROM job_queue 
        "#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)?;
    
    let jobs = jobs
        .into_iter()
        .map(Job::try_from)
        // See: https://doc.rust-lang.org/rust-by-example/error/iter_result.html#fail-the-entire-operation-with-collect
        .collect::<Result<Vec<Job>>()?;
    
    Ok(jobs)
}

impl TryFrom<QueryJob> for Job {
    type Error = crate::Error;
    
    fn try_from(j: QueryJob) -> Result<Job, crate::Error> {
        let details = JobDetails {
            id: j.id,
            locked_at: j.locked_at,
            created_at: j.created_at,
            updated_at: j.updated_at,
        };
        
        match j.r#type {
            QueryJobType::ProcessMetricsUpload => {
                #[derive(serde::Deserialize)]
                struct ProcessMetricsUploadJobData {
                    metric_name: String,
                    metric_uploaded_at: Option<DateTime<Utc>>,
                }
                
                let data: ProcessMetricsUploadJobData = serde_json::from_value(&j.data)?;
                
                Ok(Job::ProcessMetricsUpload {
                    details,
                    metric_name: data.metric_name,
                    metric_uploaded_at: data.metric_uploaded_at,
                })
            }
            QueryJobType::SendVerificationEmail => {
                #[derive(serde::Deserialize)]
                struct SendVerificationEmailJobData {
                    email_address: String,
                    email_sent_at: Option<DateTime<Utc>>,
                }
                
                let data: SendVerificationEmailJobData = serde_json::from_value(&j.data)?;
                
                Ok(Job::SendVerificationEmail {
                    details,
                    verification_email_address: data.verification_email_address,
                    verification_email_sent_at: data.verification_email_sent_at,
                })
            }
        }
    }
}
```

There are a few things you can improve here, but I omitted them for brevity. Let me know if you're interested in it and I'll create a separate HackMD doc.

#### Better

The example above is quite suboptimal in a few ways:

1. For common fields, we have to duplicate `details: JobDetails` for each enum variant. It prevents us from easily mutating common fields, because we need to match on each variant.
2. Since Rust doesn't treat enum variants as separate types ([yet!](https://github.com/rust-lang/rfcs/pull/2593)), we can't add additional functionality to specific variants (i.e. `impl Job::SendVerificationEmail`).
3. We didn't define clear boundary between "domain" types and "storage" types: even though we have inline structs like `ProcessMetricsUploadJobData` and `SendVerificationEmailJobData`, it's not clear how we can achieve `INSERT`-s.

Let's see how we can mitigate these flaws.

##### Common fields and separate functionality

We can create a "root" struct with common fields and store type-specific fields in the `JobPayload` enum:

```rust=
struct Job {
    id: String,
    status: JobStatus,
    locked_at: Option<DateTime<Utc>>,
    created_at: Option<DateTime<Utc>>,
    updated_at: Option<DateTime<Utc>>,
    
    payload: JobPayload,
}

enum JobStatus { 
    // Your statuses here...
}

enum JobPayload {
    ProcessMetricsUpload(job_payload::ProcessMetricsUpload),
    SendVerificationEmail(job_payload::SendVerificationEmail),
}

mod job_payload {
    struct ProcessMetricsUpload {
        storage_id: String,
        path: String,
        full_path: String,
    }
    
    struct SendVerificationEmail {
        email: String,
        verification_code: String,
    }
}
```

Since our `JobPayload` enum simply wraps payload-specific structs, we can do easily do this for common fields:

```rust=
impl Job {
    fn lock(&mut self) {
        self.locked_at = Some(Utc::now());
    }
}
```

And also **this!** for payload-specific fields:

```rust=
mod job_payload {
    impl ProcessMetricsUpload {
        fn replace_storage_id(&mut self, new_id: String) -> String {
            std::mem::replace(&mut self.storage_id, new_id)
        }
        
        // Any method you may need...
    }
}

impl Job {
    fn process_metrics_upload_mut(&mut self) -> Option<&mut job_payload::ProcessMetricsUpload> {
        match self.payload {
            JobPayload::ProcessMetricsUpload(payload) => Some(&mut payload),
            _ => None,
        }
    }
}

// Usage
fn process_job(job: Job) {
    if let Some(metrics) = job.process_metrics_upload_mut() {
        let old_id = metrics.replace_storage_id("123".into());
        tracing::debug!(?old_id, "Changed storage id");
    }
}
```

##### Clear mapping between "domain" and "storage" types

Now that we have `job_payload::ProcessMetricsUpload` and `job_payload::SendVerificationEmail` as separate types, we can significantly simplify conversion from `QueryJob` into `Job`:

```rust=
impl TryFrom<QueryJob> for Job {
    type Error = crate::Error;
    
    fn try_from(j: QueryJob) -> Result<Job, crate::Error> {
        // NB: The careful reader will notice that 
        //     we don't really have to match on `r#type`,
        //     but I will omit the explanation for now
        //     to make it more digestible.
        
        let payload = match j.r#type {
            QueryJobType::ProcessMetricsUpload => {                
                JobPayload::ProcessMetricsUpload(
                    serde_json::from_value(&j.payload)?
                )
            }
            QueryJobType::SendVerificationEmail => {
                JobPayload::SendVerificationEmail(
                    serde_json::from_value(&j.payload)?
                )
            }
        };
        
        Ok(Job {
            id: j.id,
            locked_at: j.locked_at,
            created_at: j.created_at,
            updated_at: j.updated_at,
            payload,
        })
    }
}
```

Each variant of `JobPayload` contains only type-specific data and we store `payload JSON(B)` column with exactly this data and nothing more or less, so conversion is really straightforward: `serde_json::Value` -> `JobPayload`.

A small nit: I don't like to perform "heavy" operations (like parsing JSON-s) inside the `try_from` method, since it hides the cost of such operation for the reader. We can just replace it with

```rust=
fn build_job(query_job: QueryJob) -> Result<Job, crate::Error> {
    // Same implementation as in `try_from`.
}
```

(A thing of preference, I guess)

Another small nit: I always keep `impl TryFrom<QueryThing> for Thing` close to `QueryThing` (and not close to `Thing`), since `Thing` shouldn't really care about `QueryThing` at all. The same applies to `impl TryFrom<Thing> for QueryThing` (I keep it near `QueryThing`).

Back to the code, we have one downside with this simplified conversion: any change to the `JobPayload` type should be considered breaking and the compiler doesn't have any means to detect such changes when calling `serde_json::from_value`. For that, we'll need a separate `QueryJobPayload` type with these goals:
1. Changes to `JobPayload` don't immediately break conversion from `serde_json::Value`, because our conversion will look like this: `serde_json::Value` (raw, Postgres level) -> `QueryJobPayload` (intermediate, "expected from Postgres" level) -> `JobPayload` (domain level).
2. `QueryJobPayload` only reflects how payload is expected to be stored in Postgres and may differ from `JobPayload` in any way until it's convertible.
3. Track breaking changes with `QueryJobPayload` exclusively, without having to worry about the rest of the application breaking.

```rust=
struct QueryJob {
    id: String,
    
    r#type: QueryJobType, // We don't really need this anymore! 
    
    payload: serde_json::Value, // We can leverage clever tricks to use `QueryJobPayload` instead of `serde_json::Value` here, 
                                // but I won't in order to keep the code simpler and easier to understand.
}

// One-to-one representation of the shape of data we expect from the JSONB column.
enum QueryJobPayload {
    ProcessMetricsUpload {
        storage_id: String,
    }
}

fn build_job(query_job: QueryJob) -> Result<Job, crate::Error> {
    // `query_job.payload: serde_json::Value` should already know which type of the job this is,
    // so we can deserialize from JSON into the `QueryJobPayload` enum without looking at `r#type`.
    let query_payload: QueryJobPayload = serde_json::from_value(&query_job.payload)?;
    
    Ok(Job {
        id: query_job.id,
        payload: query_payload.into(),
    })
}

// Any breaking change to `QueryJobPayload` will break this `impl` 
// and you'll know whether you need to adjust anything
// in the surrounding code.
impl From<QueryJobPayload> for JobPayload {    
    fn from(p: QueryJobPayload) -> JobPayload {
        match p {
            // ...
        }
    }
}
```

With this, we can do `INSERT`-s the following way:

```rust=
async fn insert_job(pg_pool: &sqlx::PgPool, job: Job) -> Result<()> {
    let query_payload: QueryJobPayload = job.payload.into();
    let payload = serde_json::to_value(&query_payload)?;
    
    sqlx::query!(
        r#"
            INSERT INTO job_queue(
                id, locked_at, payload
            )
            VALUES ($1, $2, $3),
        "#,
        job.id,
        job.locked_at,
        query_payload,
    )
    .execute(pg_pool)
    .await?;
    
    Ok(())
}

#[derive(serde::Serialize)]
enum QueryJobPayload {
    ProcessMetricsUpload {
        storage_id: String,
    }
}

// Any breaking change to `QueryJobPayload` will break the implementation here.
impl From<JobPayload> for QueryJobPayload {
    // Implementation...
}
```

### Bonus: better error handling

Suppose one of many jobs has its data corrupted. What will `get_available_jobs` return? 
```rust=
async fn get_available_jobs(&self) -> Result<Vec<Job>> { ... }
```

That's right, an error. Kinda lame that one single job results in throwing away all other jobs.

Let's change the signature a bit:
```rust=
async fn get_available_jobs(&self) -> Result<
    Vec<Result<Job, crate::Error>>,
    crate::Error,
> 
{ ... }
```

Now, we have outer `Result` which can return an error when something really unexpected happens (like Postgres is unavailable). If we managed to `SELECT` raw data (meaning rows represented by `QueryJob`), then we can return a vector of results where each result represents possible failure of converting that one respective job.

```rust=
async fn get_available_jobs(&self) -> Result<
    Vec<Result<Job, crate::Error>>,
    crate::Error,
> {
    let jobs = sqlx::query_as!(
        QueryJob,
        r#"..."#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)?;
    
    Ok(jobs
        .into_iter()
        .map(Job::try_from)
        .collect())
}

// Usage
async fn complex_routine() {
    let jobs: Vec<Job> = get_available_jobs()
        .await
        .inspect_err(|error| 
            tracing::error!(?error, "Failed to get available jobs from Postgres")
        )?
        // Now, filter out all errors and leave only 
        // successfully deserialized jobs.
        .into_iter()
        .filter_map(|result| match result {
            Ok(job) => Some(job),
            Err(error) => {
                tracing::error!(?error, "Failed to deserialize job from Postgres");
                None
            }
            
            // Or use `.ok()`:
            // result
            //     .inspect_err(|error| tracing::error!(?error, "Failed to deserialize job from Postgres"))
            //     .ok()
        })
        .collect();
}
```

Admittedly, it's a bit more hustle. Why should we even do that?
I can think of at least one reason: if we consider the application in layers, "lower level" layers should not make decisions for "higher" ones. For instance, `get_available_jobs` should not throw out valid jobs if one of them is corrupted - only "higher level" layer should decide what to do with such jobs.

In case we consider `get_available_jobs` a "higher level" layer, we can filter out and log corrupted jobs before returning them:

```rust=
async fn get_available_jobs(&self) -> Result<Vec<Job>> {
    let jobs = sqlx::query_as!(
        QueryJob,
        r#"..."#,
    )
    .fetch_all(&self.pg_pool)
    .await
    .map_err(Error::Database)?;
    
    Ok(jobs
        .into_iter()
        .filter_map(|job| match job.into() {
            Ok(job) => Some(job),
            Err(error) => {
                tracing::error!(?error, "Failed to deserialize job from Postgres");
                None
            }
        })
        .collect())
}
```