forge-jobs 0.3.1

Sidekiq-style job queue with embedded SQLite and pluggable Postgres. Per-queue workers + cron + cluster-wide rate-limit budget + cancellation that survives across replicas.
Documentation 
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

//! Backend-agnostic error type for the storage traits.
//!
//! Each backend's underlying error (sqlx, redis, postgres, …) folds
//! into a `StorageError` variant via `From` impls in that backend's
//! module. The runtime never inspects which backend produced the
//! error — it just propagates or logs.

use thiserror::Error;

#[derive(Debug, Error)]
#[non_exhaustive]
pub enum StorageError {
    #[error("schema migration failed at version {version}: {message}")]
    Migration { version: u32, message: String },

    #[error("invalid input: {0}")]
    InvalidInput(String),

    /// Runtime/process misconfiguration detected at startup (e.g. a
    /// worker started without declaring any queues). Not retryable.
    #[error("config: {0}")]
    Config(String),

    #[error("not found: {0}")]
    NotFound(String),

    #[error("conflict: {0}")]
    Conflict(String),

    #[error("io: {0}")]
    Io(#[from] std::io::Error),

    #[error("serde: {0}")]
    Serde(#[from] serde_json::Error),

    /// Generic backend-side failure. The string is whatever the
    /// driver gave us (`sqlx::Error::to_string()`, `redis::Error`, …).
    /// Use `is_transient_conflict` to decide if a retry is worth it.
    #[error("backend: {0}")]
    Backend(String),

    #[error(transparent)]
    Paths(#[from] super::paths::PathsError),
}

impl StorageError {
    /// True when this error looks like a transient MVCC / lock
    /// conflict that the caller should retry. Liberal substring
    /// matching — false positives just cost a harmless retry.
    #[must_use]
    pub fn is_transient_conflict(&self) -> bool {
        let msg = self.to_string().to_lowercase();
        msg.contains("conflict")
            || msg.contains("database is locked")
            || msg.contains("sqlite_busy")
            || msg.contains("deadlock detected")
            || msg.contains("could not serialize access")
            || msg.contains("write conflict")
            || msg.contains("transaction was aborted")
            // sqlx pool-acquire timeout under a backed-up single writer:
            // the work isn't lost, the writer queue is just saturated —
            // retrying once it drains is the right move.
            || msg.contains("pool timed out")
            || msg.contains("timed out while waiting")
    }
}

pub type Result<T> = std::result::Result<T, StorageError>;