rust-job-queue-api-worker-system 0.1.0

A production-shaped Rust job queue: Axum API + async workers + Postgres SKIP LOCKED dequeue, retries with decorrelated jitter, idempotency, cooperative cancellation, OpenAPI, Prometheus metrics.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

//! The crate's error taxonomy.
//!
//! `JobError` is the error type returned by every fallible function in
//! `crate::queue`, `crate::db`, and `crate::payload`. The HTTP layer in
//! `crate::api::error` maps each variant to a specific HTTP status; the
//! worker layer logs them. Keeping the taxonomy flat (one enum, six
//! variants) means there is exactly one place to update when a new
//! failure mode is introduced.

use thiserror::Error;

use crate::ids::JobId;

/// Every fallible queue operation returns this. Variants are mapped to
/// HTTP statuses in [`crate::api::error::ApiError`]; worker callers
/// typically log and continue.
#[derive(Debug, Error)]
pub enum JobError {
    /// No row matches the given id. Maps to HTTP 404.
    #[error("job {0} not found")]
    NotFound(JobId),

    /// The requested state transition is not legal for the current
    /// status of the row (e.g., succeed-on-cancelled, fail-on-queued).
    /// Maps to HTTP 409 (Conflict).
    #[error("invalid state transition for job {id}: cannot {action} when status={status}")]
    InvalidTransition {
        id: JobId,
        status: String,
        action: &'static str,
    },

    /// The payload JSON did not match the schema for the requested
    /// kind. The `reason` field carries the serde error message —
    /// usually a clear "missing field" / "expected type" string.
    /// Maps to HTTP 422.
    #[error("payload validation failed for {kind}: {reason}")]
    PayloadInvalid { kind: &'static str, reason: String },

    /// The partial unique index matched on `idempotency_key` but the
    /// follow-up lookup returned no row. Not expected in normal use;
    /// would require the existing row to be deleted between the
    /// `INSERT ... ON CONFLICT` and the `SELECT`. Maps to HTTP 409.
    #[error("idempotency conflict on key {0}")]
    IdempotencyConflict(String),

    /// Any underlying Postgres / sqlx error. Maps to HTTP 500 (with a
    /// redacted body; full detail goes to tracing).
    #[error(transparent)]
    Db(#[from] sqlx::Error),

    /// Any underlying JSON serialisation / deserialisation error from
    /// payload handling. Maps to HTTP 422.
    #[error(transparent)]
    Serde(#[from] serde_json::Error),
}

impl JobError {
    /// A short snake_case label for this error, suitable for use as a
    /// metric label, a log field, or the `error` slug in an HTTP
    /// response body. Allocation-free.
    pub fn kind(&self) -> &'static str {
        match self {
            JobError::NotFound(_) => "not_found",
            JobError::InvalidTransition { .. } => "invalid_transition",
            JobError::PayloadInvalid { .. } => "payload_invalid",
            JobError::IdempotencyConflict(_) => "idempotency_conflict",
            JobError::Db(_) => "db",
            JobError::Serde(_) => "serde",
        }
    }
}