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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
use crate::{registry::get_job_handler, config::is_shutting_down};
use crate::utils::rdconfig::{get_redis_connection, get_dedicated_connection};
use tokio::time::{sleep, Duration};
use redis::AsyncCommands;
use chrono::Utc;
use crate::utils::constants::{DELAYED_JOBS_KEY, MAX_RETRIES};
/// How long a worker blocks on BLMOVE before looping to re-check shutdown.
const BLOCK_TIMEOUT_SECS: f64 = 5.0;
/// Requeue any job ids left in the per-queue in-flight list by a previous run that
/// crashed mid-processing, so they are retried instead of lost.
///
/// Note: this assumes a single engine process per queue. If you run multiple engine
/// processes sharing one Redis, a restart of one may requeue jobs another is actively
/// processing (at-least-once semantics — handlers should be idempotent).
async fn recover_inflight_jobs(queue: &str) {
let inflight_key = format!("snm:inflight:{}", queue);
let queue_key = format!("snm:queue:{}", queue);
let mut conn = match get_redis_connection().await {
Ok(c) => c,
Err(_) => return,
};
let mut recovered = 0u64;
loop {
// Move oldest in-flight id back to the head of the queue.
let moved: Option<String> = conn
.lmove(&inflight_key, &queue_key, redis::Direction::Right, redis::Direction::Left)
.await
.unwrap_or(None);
if moved.is_none() {
break;
}
recovered += 1;
}
if recovered > 0 {
tracing::warn!("Recovered {} orphaned in-flight job(s) for queue '{}'", recovered, queue);
}
}
pub async fn start_worker_pool(queue: &str, concurrency: usize) {
recover_inflight_jobs(queue).await;
for _ in 0..concurrency {
let queue = queue.to_string();
tokio::spawn(async move {
let queue_key = format!("snm:queue:{}", queue);
let inflight_key = format!("snm:inflight:{}", queue);
// Dedicated connection for the blocking BLMOVE fetch, so it never stalls
// the shared connection used for enqueue/metrics/result writes.
let mut fetch_conn = match get_dedicated_connection().await {
Ok(c) => c,
Err(_) => return,
};
loop {
if is_shutting_down() {
break;
}
// Reliable fetch: atomically move the job id from the queue head into a
// per-queue in-flight list, so a crash mid-processing doesn't lose the job.
// BLMOVE blocks up to BLOCK_TIMEOUT_SECS instead of busy-polling; a short
// timeout lets the loop periodically re-check the shutdown flag.
let job_id: Option<String> = match fetch_conn
.blmove(
&queue_key,
&inflight_key,
redis::Direction::Left,
redis::Direction::Right,
BLOCK_TIMEOUT_SECS,
)
.await
{
Ok(v) => v,
Err(_) => {
// Connection hiccup: back off briefly and reconnect.
sleep(Duration::from_secs(1)).await;
if let Ok(c) = get_dedicated_connection().await {
fetch_conn = c;
}
continue;
}
};
let Some(job_id) = job_id else { continue };
let mut conn = match get_redis_connection().await {
Ok(c) => c,
Err(_) => {
sleep(Duration::from_secs(1)).await;
continue;
}
};
{
// Recompute the date each job so a long-lived worker rolls over to
// the new day's stats keys instead of writing to its start date forever.
let today = Utc::now().date_naive().format("%Y-%m-%d").to_string();
let job_key = format!("snm:job:{}", job_id);
let job_payload: String = conn.hget(&job_key, "payload").await.unwrap_or_default();
let job_name: String = conn.hget(&job_key, "job_name").await.unwrap_or_default();
let mut handled = false;
let mut error_message: Option<String> = None;
// Dispatch to the specific handler registered under this job's name,
// instead of guessing by trying every registered handler.
match get_job_handler(&job_name) {
Some(handler) => match handler(job_payload.clone()).await {
Ok(job) => {
if job.before().await.is_err() {
// Skipped is a terminal, non-failure state.
let _: () = conn.hset_multiple(&job_key, &[
("status", "skipped"),
("skipped_at", &Utc::now().to_rfc3339()),
]).await.unwrap_or_default();
} else {
match job.perform().await {
Ok(_) => {
let _ = job.after().await;
let _: () = conn.hset_multiple(&job_key, &[
("status", "success"),
("completed_at", &Utc::now().to_rfc3339()),
]).await.unwrap_or_default();
let _: () = conn.incr("snm:qrush:success", 1).await.unwrap_or_default();
// Track success jobs
let _: () = conn.rpush(format!("snm:success:{}", queue), &job_id).await.unwrap_or_default();
let key = format!("snm:stats:jobs:{}", today);
// increment daily success counter
let _: () = conn.incr(&key, 1).await.unwrap_or_default();
let _: () = conn.incr("snm:qrush:total_jobs", 1).await.unwrap_or_default();
}
Err(err) => {
let _ = job.on_error(&err).await;
let retries: i64 = conn.hincr(&job_key, "retries", 1).await.unwrap_or(1);
let _: () = conn.hset(&job_key, "error", err.to_string()).await.unwrap_or_default();
if retries <= MAX_RETRIES as i64 {
// Exponential backoff (10s * 2^retries) plus jitter, to avoid
// thundering-herd retries. Jitter is derived from the current
// sub-second clock, so no extra RNG dependency is needed.
let base = 10_i64 * (1_i64 << retries.min(16));
let jitter = (Utc::now().timestamp_subsec_nanos() as i64) % base.max(1);
let run_at = Utc::now().timestamp() + base + jitter;
let _: () = conn.hset_multiple(&job_key, &[
("status", "retrying".to_string()),
("run_at", run_at.to_string()),
]).await.unwrap_or_default();
let _: () = conn.rpush(format!("snm:retry:{}", queue), &job_id).await.unwrap_or_default();
let _: () = conn.zadd(DELAYED_JOBS_KEY, &job_id, run_at).await.unwrap_or_default();
} else {
// Retries exhausted: move to the dead-letter queue instead of
// silently dropping the job.
let _: () = conn.hset_multiple(&job_key, &[
("status", "dead".to_string()),
("dead_at", Utc::now().to_rfc3339()),
]).await.unwrap_or_default();
let _: () = conn.rpush("snm:dead_jobs", &job_id).await.unwrap_or_default();
let _: () = conn.rpush(format!("snm:dead:{}", queue), &job_id).await.unwrap_or_default();
let _: () = conn.incr("snm:qrush:dead", 1).await.unwrap_or_default();
}
}
}
let _ = job.always().await;
}
handled = true;
}
Err(e) => {
error_message = Some(e.to_string());
}
},
None => {
error_message = Some(format!("No handler registered for job '{}'", job_name));
}
}
if !handled {
let mut hset_data = vec![
("status", "failed".to_string()),
("failed_at", Utc::now().to_rfc3339()),
("queue", queue.clone()),
("failed_at", Utc::now().to_rfc3339()),
];
if let Some(ref emsg) = error_message {
hset_data.push(("error", emsg.clone()));
}
let fail_key = format!("snm:stats:jobs:{}:failed", today);
// increment daily failed counter
let _: () = conn.incr(&fail_key, 1).await.unwrap_or_default();
let _: () = conn.hset_multiple(&job_key, &hset_data).await.unwrap_or_default();
let _: Result<(), _> = conn.lpush(
format!("snm:logs:{}", queue),
format!("[{}] ❌ Job {} failed", Utc::now(), job_id),
).await;
let _: Result<(), _> = conn.ltrim(format!("snm:logs:{}", queue), 0, 99).await;
let _: () = conn.rpush("snm:failed_jobs", &job_id).await.unwrap_or_default();
// Track failed jobs
let _: () = conn.rpush(format!("snm:failed:{}", queue), &job_id).await.unwrap_or_default();
let _: () = conn.incr("snm:qrush:failed", 1).await.unwrap_or_default();
}
// The job has reached a terminal state (success / skipped / failed) or
// has been scheduled for retry, so it is no longer in flight.
let _: () = conn.lrem(&inflight_key, 1, &job_id).await.unwrap_or_default();
}
}
});
}
}
pub async fn start_delayed_worker_pool() {
tokio::spawn(async move {
loop {
if is_shutting_down() {
break;
}
let now = chrono::Utc::now().timestamp();
let mut conn = match get_redis_connection().await {
Ok(c) => c,
Err(_) => {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
};
let jobs: Vec<String> = conn.zrangebyscore(DELAYED_JOBS_KEY, 0, now).await.unwrap_or_default();
for job_id in jobs {
// Atomic claim: ZREM removes the member exactly once, so only the worker
// whose ZREM returns 1 enqueues it — no duplicates across processes.
let claimed: i64 = conn.zrem(DELAYED_JOBS_KEY, &job_id).await.unwrap_or(0);
if claimed == 0 {
continue;
}
// Route back to the job's original queue instead of hardcoding "default".
let job_key = format!("snm:job:{}", job_id);
let queue: String = conn.hget(&job_key, "queue").await.unwrap_or_default();
let queue = if queue.is_empty() { "default".to_string() } else { queue };
let _: () = conn.hset(&job_key, "status", "pending").await.unwrap_or_default();
let _: () = conn.rpush(format!("snm:queue:{}", queue), &job_id).await.unwrap_or_default();
}
tokio::time::sleep(Duration::from_secs(5)).await;
}
});
}