arcature 0.1.0

Arcature: an opinionated full-stack Rust web framework. One package, batteries included.
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

//! SQLite statement text.
//!
//! Placeholders are `?`, bound in order of appearance. Timestamps are epoch
//! milliseconds in `INTEGER` columns, so "now" is spelled
//! `CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)` -- 2440587.5 is
//! the Julian day of the Unix epoch. SQLite has no `now()`, and
//! `CURRENT_TIMESTAMP` yields text whose format does not compare correctly
//! against anything else the queue writes. The expression is spelled out at
//! every use because `concat!` cannot splice a `const`, and building the
//! statement at runtime would give up the `&'static str` SQL rule.

/// Every statement the queue issues against SQLite.
pub(crate) mod sql {
    /// Wait rather than fail when another connection holds the write lock.
    /// A claim transaction is a couple of round trips, so a claimer that
    /// waits is waiting on another claimer, never on a job running.
    pub(crate) const SESSION_SETUP: Option<&str> = Some("PRAGMA busy_timeout = 5000");

    /// Take the write lock up front. A deferred `BEGIN` upgrades to a write
    /// lock only at the first write, which can fail with `SQLITE_BUSY` after
    /// the read has already happened -- exactly the window in which two
    /// claimers could pick the same row.
    pub(crate) const BEGIN: &str = "BEGIN IMMEDIATE";

    /// Pick claimable rows. No locking clause: the enclosing `BEGIN
    /// IMMEDIATE` already excludes every other writer, so there is no other
    /// claimer's lock to skip.
    /// Binds: batch size.
    pub(crate) const CLAIM_PICK: &str = r#"SELECT id, kind, version, payload, attempts, max_attempts
  FROM arcature_jobs
 WHERE status = 'pending'
   AND available_at <= CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 ORDER BY available_at, id
 LIMIT ?"#;

    /// Mark one picked row as claimed.
    /// Binds: worker id, claim token, lease seconds, job id.
    pub(crate) const CLAIM_MARK: &str = r#"UPDATE arcature_jobs
   SET status = 'running', attempts = attempts + 1, locked_by = ?,
       locked_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       claim_token = ?, lease_seconds = ?, last_error = NULL,
       last_error_kind = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'pending'"#;

    /// Insert one job row. Binds: id, kind, version, payload, max attempts,
    /// run at, available at.
    pub(crate) const INSERT: &str = r#"INSERT INTO arcature_jobs
       (id, kind, version, payload, max_attempts, run_at, available_at)
VALUES (?, ?, ?, ?, ?, ?, ?)"#;

    /// Binds: job id, claim token.
    pub(crate) const MARK_SUCCEEDED: &str = r#"UPDATE arcature_jobs
   SET status = 'succeeded', locked_by = NULL, locked_at = NULL,
       claim_token = NULL, last_error = NULL, last_error_kind = NULL,
       failed_at = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'running' AND claim_token = ?"#;

    /// Binds: available at, message, job id, claim token.
    pub(crate) const MARK_RETRY: &str = r#"UPDATE arcature_jobs
   SET status = 'pending', available_at = ?, locked_by = NULL,
       locked_at = NULL, claim_token = NULL, last_error = ?,
       last_error_kind = 'retryable', failed_at = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'running' AND claim_token = ?"#;

    /// Binds: message, error kind, job id, claim token.
    pub(crate) const MARK_DEAD: &str = r#"UPDATE arcature_jobs
   SET status = 'dead', locked_by = NULL, locked_at = NULL,
       claim_token = NULL, last_error = ?, last_error_kind = ?,
       failed_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'running' AND claim_token = ?"#;

    /// Binds: lease seconds, job id, claim token.
    pub(crate) const HEARTBEAT: &str = r#"UPDATE arcature_jobs
   SET locked_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       lease_seconds = ?,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'running' AND claim_token = ?"#;

    /// Binds: job id, claim token.
    pub(crate) const RELEASE_CLAIM: &str = r#"UPDATE arcature_jobs
   SET status = 'pending',
       available_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       locked_by = NULL, locked_at = NULL, claim_token = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'running' AND claim_token = ?"#;

    /// Binds: batch size.
    pub(crate) const SWEEP_DEAD: &str = r#"UPDATE arcature_jobs
   SET status = 'dead', locked_by = NULL, locked_at = NULL,
       claim_token = NULL,
       last_error = 'job exhausted its max_attempts after a crash',
       last_error_kind = 'exhausted',
       failed_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id IN (
     SELECT id FROM arcature_jobs
      WHERE status = 'running'
        AND locked_at IS NOT NULL
        AND locked_at + lease_seconds * 1000
            < CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
        AND attempts >= max_attempts
      LIMIT ?
 )"#;

    /// Binds: batch size.
    pub(crate) const SWEEP_REQUEUE: &str = r#"UPDATE arcature_jobs
   SET status = 'pending',
       available_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       locked_by = NULL, locked_at = NULL, claim_token = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id IN (
     SELECT id FROM arcature_jobs
      WHERE status = 'running'
        AND locked_at IS NOT NULL
        AND locked_at + lease_seconds * 1000
            < CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
        AND attempts < max_attempts
      LIMIT ?
 )"#;

    /// Binds: job id.
    pub(crate) const CANCEL: &str = r#"UPDATE arcature_jobs
   SET status = 'cancelled', locked_by = NULL, locked_at = NULL,
       claim_token = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status IN ('pending', 'running')"#;

    /// Binds: job id.
    pub(crate) const REQUEUE_DEAD: &str = r#"UPDATE arcature_jobs
   SET status = 'pending', attempts = 0,
       available_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER),
       locked_by = NULL, locked_at = NULL, claim_token = NULL,
       last_error = NULL, last_error_kind = NULL, failed_at = NULL,
       updated_at = CAST((julianday('now')-2440587.5)*86400000 AS INTEGER)
 WHERE id = ? AND status = 'dead'"#;

    /// The migration history table.
    pub(crate) const CREATE_HISTORY: &str = r#"CREATE TABLE IF NOT EXISTS arcature_jobs_schema_migrations (
    version    TEXT PRIMARY KEY,
    applied_at INTEGER NOT NULL
        DEFAULT (CAST((julianday('now')-2440587.5)*86400000 AS INTEGER))
)"#;

    /// Binds: version.
    pub(crate) const COUNT_APPLIED: &str =
        "SELECT COUNT(*) FROM arcature_jobs_schema_migrations WHERE version = ?";

    /// Binds: version. Idempotent so a racing migrator cannot fail on the
    /// primary key.
    pub(crate) const RECORD_APPLIED: &str =
        "INSERT OR IGNORE INTO arcature_jobs_schema_migrations (version) VALUES (?)";

    /// SQLite has no advisory lock, and needs none here: every statement in
    /// the migration is idempotent (`IF NOT EXISTS`, `INSERT OR IGNORE`) and
    /// SQLite serialises writers anyway, so two migrators racing converge on
    /// the same schema instead of conflicting.
    pub(crate) const LOCK: Option<&str> = None;

    /// See [`LOCK`].
    pub(crate) const UNLOCK: Option<&str> = None;

    /// The schema, one statement per `--;;` separated chunk.
    pub(crate) const SCHEMA: &str = include_str!("../migrations/sqlite/0001_jobs.sql");
}