GITGUD

neiam neiam /dms

public · Issues · Pulls · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

android background worker

7bac2a7 · gmorell · 2026-05-09 15:46

15 files +278 -20

Files changed

modified README.md
+1 −0
@@ -110,6 +110,7 @@ All env-driven — defaults make sense for local dev.
110 110 | `DMS_PUBLIC_URL` | (unset) | Override what the QR pairing URL encodes when behind a proxy. Falls back to the request's Host header, then the bind address. |
111 111 | `WORKER_KIND` | (required for workers) | `thumbnail` / `ocr` / `describe` / `extract_tags`. Same as `--kind`. |
112 112 | `WORKER_PREFETCH` | `4` | Per-worker concurrency (RabbitMQ basic_qos). Tune CPU-bound (OCR) lower than I/O-bound. |
113 +| `WORKER_MAX_ATTEMPTS` | `100` | Hard cap on per-(upload, kind) DB-tracked attempts. Once exceeded, the worker DLQs the next redelivery instead of running it — protects the queue from a single poisoned message hogging consumers. Set `0` to disable. |
113 114 | `OCRS_DETECTION_MODEL` | `/opt/dms/models/text-detection.rten` | Bundled in the image; override to swap models. Required for OCR — outside Docker (e.g. `cargo run`), set this to a downloaded `.rten` file or the worker DLQs jobs with a clear log line. |
114 115 | `OCRS_RECOGNITION_MODEL` | `/opt/dms/models/text-recognition.rten` | Same as above. |
115 116
modified src/bin/webservice.rs
+4 −0
@@ -83,6 +83,8 @@ async fn main() -> Result<(), anyhow::Error> {
83 83 .route("/search", get(handlers::search_uploads))
84 84 .route("/failed", get(handlers::show_failed))
85 85 .route("/failed/retry", post(handlers::retry_all_failed))
86 + .route("/failed/dlq/requeue", post(handlers::requeue_all_dlqs_html))
87 + .route("/failed/dlq/{kind}/requeue", post(handlers::requeue_dlq_kind_html))
86 88 .route("/uploads/{id}/jobs/{kind}/retry", post(handlers::retry_one))
87 89 .layer(DefaultBodyLimit::max(config.max_upload_bytes))
88 90 .layer(axum::middleware::from_fn_with_state(
@@ -98,6 +100,8 @@ async fn main() -> Result<(), anyhow::Error> {
98 100 .route("/search", get(handlers::api_search_uploads))
99 101 .route("/failed", get(handlers::api_list_failed))
100 102 .route("/failed/retry", post(handlers::api_retry_all_failed))
103 + .route("/failed/dlq/requeue", post(handlers::api_requeue_all_dlqs))
104 + .route("/failed/dlq/{kind}/requeue", post(handlers::api_requeue_dlq_kind))
101 105 .layer(DefaultBodyLimit::max(config.max_upload_bytes))
102 106 .layer(axum::middleware::from_fn_with_state(
103 107 app_state.db.clone(),
modified src/config.rs
+5 −0
@@ -10,6 +10,7 @@ pub struct Config {
10 10 pub storage_root: String,
11 11 pub amqp_url: String,
12 12 pub worker_prefetch: u16,
13 + pub worker_max_attempts: i32,
13 14 pub max_upload_bytes: usize,
14 15 pub auto_migrate: bool,
15 16 pub registration_enabled: bool,
@@ -55,6 +56,10 @@ impl Config {
55 56 .unwrap_or_else(|_| "4".to_string())
56 57 .parse()
57 58 .unwrap_or(4),
59 + worker_max_attempts: env::var("WORKER_MAX_ATTEMPTS")
60 + .ok()
61 + .and_then(|v| v.parse().ok())
62 + .unwrap_or(100),
58 63 max_upload_bytes: env::var("MAX_UPLOAD_BYTES")
59 64 .unwrap_or_else(|_| "104857600".to_string()) // 100 MiB
60 65 .parse()
modified src/handlers/failed.rs
+82 −1
@@ -20,7 +20,9 @@ use uuid::Uuid;
20 20 use crate::entities::{upload_jobs, uploads};
21 21 use crate::jobs;
22 22 use crate::middleware::UserContext;
23 −use crate::queue::{publish_job, queue_depths, JobKind, JobMessage, QueueDepth};
23 +use crate::queue::{
24 + publish_job, queue_depths, requeue_all_dlqs, requeue_dlq, JobKind, JobMessage, QueueDepth,
25 +};
24 26 use crate::AppState;
25 27
26 28 #[derive(Debug, Serialize, Clone)]
@@ -292,6 +294,85 @@ pub async fn retry_one(
292 294 Ok(Redirect::to("/failed?notice=Retry+queued."))
293 295 }
294 296
297 +/// Drain every per-kind DLQ back into its main queue and redirect to the
298 +/// Work page. Operates broker-globally (DLQs aren't user-scoped — neither
299 +/// is the depth display).
300 +pub async fn requeue_all_dlqs_html(
301 + State(state): State<AppState>,
302 + _user: UserContext,
303 +) -> Result<Redirect, StatusCode> {
304 + let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
305 + error!(error = %e, "amqp connect failed");
306 + StatusCode::INTERNAL_SERVER_ERROR
307 + })?;
308 + let n = requeue_all_dlqs(&conn).await.map_err(|e| {
309 + error!(error = %e, "requeue_all_dlqs failed");
310 + StatusCode::INTERNAL_SERVER_ERROR
311 + })?;
312 + info!(moved = n, "requeue all DLQs");
313 + let msg = if n == 0 {
314 + "No DLQ messages to requeue.".to_string()
315 + } else {
316 + format!("Requeued {n} DLQ message(s).")
317 + };
318 + Ok(Redirect::to(&format!("/failed?notice={}", urlencode(&msg))))
319 +}
320 +
321 +pub async fn api_requeue_all_dlqs(
322 + State(state): State<AppState>,
323 + _user: UserContext,
324 +) -> Result<Json<serde_json::Value>, StatusCode> {
325 + let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
326 + error!(error = %e, "amqp connect failed");
327 + StatusCode::INTERNAL_SERVER_ERROR
328 + })?;
329 + let n = requeue_all_dlqs(&conn).await.map_err(|e| {
330 + error!(error = %e, "requeue_all_dlqs failed");
331 + StatusCode::INTERNAL_SERVER_ERROR
332 + })?;
333 + Ok(Json(serde_json::json!({ "requeued": n })))
334 +}
335 +
336 +pub async fn requeue_dlq_kind_html(
337 + State(state): State<AppState>,
338 + _user: UserContext,
339 + Path(kind_str): Path<String>,
340 +) -> Result<Redirect, StatusCode> {
341 + let kind = JobKind::from_str(&kind_str).ok_or(StatusCode::BAD_REQUEST)?;
342 + let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
343 + error!(error = %e, "amqp connect failed");
344 + StatusCode::INTERNAL_SERVER_ERROR
345 + })?;
346 + let n = requeue_dlq(&conn, kind).await.map_err(|e| {
347 + error!(error = %e, kind = ?kind, "requeue_dlq failed");
348 + StatusCode::INTERNAL_SERVER_ERROR
349 + })?;
350 + info!(moved = n, kind = ?kind, "requeue one DLQ");
351 + let msg = if n == 0 {
352 + format!("No DLQ messages for {} to requeue.", kind.as_str())
353 + } else {
354 + format!("Requeued {n} {} DLQ message(s).", kind.as_str())
355 + };
356 + Ok(Redirect::to(&format!("/failed?notice={}", urlencode(&msg))))
357 +}
358 +
359 +pub async fn api_requeue_dlq_kind(
360 + State(state): State<AppState>,
361 + _user: UserContext,
362 + Path(kind_str): Path<String>,
363 +) -> Result<Json<serde_json::Value>, StatusCode> {
364 + let kind = JobKind::from_str(&kind_str).ok_or(StatusCode::BAD_REQUEST)?;
365 + let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
366 + error!(error = %e, "amqp connect failed");
367 + StatusCode::INTERNAL_SERVER_ERROR
368 + })?;
369 + let n = requeue_dlq(&conn, kind).await.map_err(|e| {
370 + error!(error = %e, kind = ?kind, "requeue_dlq failed");
371 + StatusCode::INTERNAL_SERVER_ERROR
372 + })?;
373 + Ok(Json(serde_json::json!({ "kind": kind.as_str(), "requeued": n })))
374 +}
375 +
295 376 fn urlencode(s: &str) -> String {
296 377 s.bytes()
297 378 .map(|b| match b {
modified src/handlers/mod.rs
+2 −0
@@ -9,6 +9,8 @@ pub use auth::{show_invites, handle_create_invite};
9 9 pub use failed::{
10 10 show_failed, retry_all_failed, retry_one,
11 11 api_list_failed, api_retry_all_failed,
12 + requeue_all_dlqs_html, requeue_dlq_kind_html,
13 + api_requeue_all_dlqs, api_requeue_dlq_kind,
12 14 };
13 15 pub use uploads::{
14 16 list_uploads, upload_file, show_upload, enqueue_job, add_tags, search_uploads,
modified src/queue/mod.rs
+50 −0
@@ -7,6 +7,7 @@ pub use publisher::{publish_job, Publisher};
7 7 pub use consumer::{run_worker, JobHandler, JobOutcome};
8 8
9 9 use lapin::{Connection, ConnectionProperties};
10 +use lapin::options::{BasicAckOptions, BasicGetOptions, BasicPublishOptions};
10 11
11 12 pub const EXCHANGE: &str = "dms.jobs";
12 13 pub const DLX_EXCHANGE: &str = "dms.jobs.dlx";
@@ -91,3 +92,52 @@ pub async fn queue_depths(conn: &Connection) -> Result<Vec<QueueDepth>, lapin::E
91 92 }
92 93 Ok(out)
93 94 }
95 +
96 +/// Drain the DLQ for a single job kind back into the main per-kind queue.
97 +/// Each message is fetched with `basic_get` and re-published to `EXCHANGE`
98 +/// using the kind's routing key, preserving the original payload + props
99 +/// (including the accumulated `x-death` header — useful for diagnostics).
100 +///
101 +/// Returns the number of messages moved. Stops when the DLQ is empty or when
102 +/// a re-publish fails (the corresponding DLQ message is left un-acked so the
103 +/// broker requeues it).
104 +pub async fn requeue_dlq(conn: &Connection, kind: JobKind) -> Result<usize, lapin::Error> {
105 + let ch = conn.create_channel().await?;
106 + let dlq = dlq_name(kind);
107 + let rk = routing_key(kind);
108 + let mut moved = 0usize;
109 + loop {
110 + let got = ch
111 + .basic_get(&dlq, BasicGetOptions { no_ack: false })
112 + .await?;
113 + let Some(msg) = got else { break };
114 + ch.basic_publish(
115 + EXCHANGE,
116 + rk,
117 + BasicPublishOptions::default(),
118 + &msg.data,
119 + msg.properties.clone(),
120 + )
121 + .await?
122 + .await?;
123 + ch.basic_ack(msg.delivery_tag, BasicAckOptions::default()).await?;
124 + moved += 1;
125 + }
126 + Ok(moved)
127 +}
128 +
129 +/// Convenience: requeue every kind's DLQ. Returns the total moved across all
130 +/// kinds. A failure on one kind aborts the run — the caller can retry.
131 +pub async fn requeue_all_dlqs(conn: &Connection) -> Result<usize, lapin::Error> {
132 + let mut total = 0usize;
133 + for kind in [
134 + JobKind::Thumbnail,
135 + JobKind::Ocr,
136 + JobKind::Describe,
137 + JobKind::ExtractTags,
138 + JobKind::SimilaritySearch,
139 + ] {
140 + total += requeue_dlq(conn, kind).await?;
141 + }
142 + Ok(total)
143 +}
modified src/workers/mod.rs
+48 −2
@@ -4,10 +4,13 @@ pub mod describe;
4 4 pub mod extract_tags;
5 5
6 6 use async_trait::async_trait;
7 −use sea_orm::DatabaseConnection;
7 +use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter};
8 8 use std::sync::Arc;
9 +use tracing::{error, warn};
9 10
10 −use crate::queue::{JobHandler, JobMessage, JobOutcome};
11 +use crate::entities::upload_jobs;
12 +use crate::jobs;
13 +use crate::queue::{JobHandler, JobKind, JobMessage, JobOutcome};
11 14 use crate::storage::LocalStorage;
12 15 use crate::Config;
13 16
@@ -36,6 +39,49 @@ pub struct DispatchHandler {
36 39 #[async_trait]
37 40 impl JobHandler for DispatchHandler {
38 41 async fn handle(&self, msg: &JobMessage) -> JobOutcome {
42 + // Hard cap on retries. The DB-tracked `upload_jobs.attempts` column
43 + // ticks up every time `mark_processing` runs, so a worker stuck in a
44 + // requeue loop can build up arbitrarily many redeliveries; without a
45 + // cap, the same poisoned message starves the queue forever.
46 + let max = self.ctx.config.worker_max_attempts;
47 + if max > 0 {
48 + let kind = msg.kind;
49 + match read_attempts(&self.ctx.db, msg.upload_id, kind).await {
50 + Ok(Some(prior)) if prior >= max => {
51 + let err_msg = format!(
52 + "exceeded WORKER_MAX_ATTEMPTS ({max}); giving up after {prior} attempts"
53 + );
54 + warn!(
55 + upload_id = %msg.upload_id,
56 + kind = ?kind,
57 + attempts = prior,
58 + max,
59 + "max attempts exceeded, sending to DLQ"
60 + );
61 + if let Err(e) = jobs::mark_failed(self.ctx.db.as_ref(), msg.upload_id, kind, &err_msg).await {
62 + error!(error = %e, "mark_failed during attempts-cap reject");
63 + }
64 + return JobOutcome::Reject;
65 + }
66 + Ok(_) => {}
67 + Err(e) => {
68 + error!(error = %e, "read_attempts failed; allowing job through");
69 + }
70 + }
71 + }
39 72 (self.run)(self.ctx.clone(), msg.clone()).await
40 73 }
41 74 }
75 +
76 +async fn read_attempts(
77 + db: &DatabaseConnection,
78 + upload_id: uuid::Uuid,
79 + kind: JobKind,
80 +) -> Result<Option<i32>, sea_orm::DbErr> {
81 + let row = upload_jobs::Entity::find()
82 + .filter(upload_jobs::Column::UploadId.eq(upload_id))
83 + .filter(upload_jobs::Column::Kind.eq(kind.as_str()))
84 + .one(db)
85 + .await?;
86 + Ok(row.map(|r| r.attempts))
87 +}
modified templates/_index.html
+61 −16
@@ -5,17 +5,21 @@
5 5 Stash images. Find them with a phrase.
6 6 </h1>
7 7 <p class="text-base-content/70 text-lg sm:text-xl mt-6 max-w-2xl mx-auto leading-relaxed">
8 − OCR every screenshot, describe every photo, tag everything as it lands.
9 − Search later by anything a categorizer noticed — "McMaster catalog cover"
10 − finds the photo of the McMaster catalog cover.
8 + Self-hosted Dank Macro Stash for screenshots and photos. Every upload
9 + fans out to a queue of categorizers — thumbnails, OCR, image description,
10 + tag extraction, similarity. Anything any pipeline notices becomes a tag,
11 + so a year from now "McMaster catalog cover" still finds it.
11 12 </p>
12 − <div class="flex justify-center gap-3 mt-8">
13 + <div class="flex justify-center gap-3 mt-8 flex-wrap">
13 14 {% if user %}
14 15 <a href="/uploads" class="btn btn-primary">
15 16 <i class="fas fa-fw fa-images"></i> Open uploads
16 17 </a>
17 − <a href="/invites" class="btn btn-ghost">
18 − <i class="fas fa-fw fa-envelope"></i> Invites
18 + <a href="/search" class="btn btn-ghost">
19 + <i class="fas fa-fw fa-magnifying-glass"></i> Search
20 + </a>
21 + <a href="/api-tokens" class="btn btn-ghost">
22 + <i class="fas fa-fw fa-mobile-screen"></i> Pair a device
19 23 </a>
20 24 {% else %}
21 25 {% if registration_open %}
@@ -42,12 +46,12 @@
42 46 <div class="border border-base-300 rounded-box p-5 bg-base-200">
43 47 <div class="flex items-center gap-2 mb-2">
44 48 <i class="fas fa-fw fa-bolt size-5 text-accent"></i>
45 − <h3 class="font-semibold">Auto-categorizers</h3>
49 + <h3 class="font-semibold">Five auto-categorizers</h3>
46 50 </div>
47 51 <p class="text-sm text-base-content/80">
48 − Every upload is fanned out to a queue: OCR, image describe (VLM),
49 − similar-image detection, tag extraction. Each pipeline can be re-run
50 − on demand from the detail page.
52 + Every upload fans out to thumbnail, OCR (<a href="https://github.com/robertknight/ocrs" class="link">ocrs</a>),
53 + image describe (VLM), tag extraction, and similarity-search. Each
54 + pipeline can be re-run on demand from the detail page.
51 55 </p>
52 56 </div>
53 57
@@ -71,7 +75,21 @@
71 75 <p class="text-sm text-base-content/80">
72 76 Search by file name, item description, or any tag from any pipeline.
73 77 OCR text becomes <code>ocr:*</code> tags, descriptions become
74 − <code>desc:*</code> — the same query box hits all of them.
78 + <code>desc:*</code> — toggle sources per query, or use <code>ns:*</code>
79 + to enumerate every value in a namespace.
80 + </p>
81 + </div>
82 +
83 + <div class="border border-base-300 rounded-box p-5 bg-base-200">
84 + <div class="flex items-center gap-2 mb-2">
85 + <i class="fas fa-fw fa-mobile-screen size-5 text-accent"></i>
86 + <h3 class="font-semibold">Android companion</h3>
87 + </div>
88 + <p class="text-sm text-base-content/80">
89 + QR-pair from <a href="/api-tokens" class="link">/api-tokens</a> and
90 + you're in. Share images straight from the gallery into DMS, search and
91 + retag from your pocket, share back out via the system sheet. Uploads
92 + run in a foreground service — fire and forget.
75 93 </p>
76 94 </div>
77 95
@@ -99,6 +117,19 @@
99 117 </p>
100 118 </div>
101 119
120 + <div class="border border-base-300 rounded-box p-5 bg-base-200">
121 + <div class="flex items-center gap-2 mb-2">
122 + <i class="fas fa-fw fa-gear fa-spin size-5 text-accent"></i>
123 + <h3 class="font-semibold">Operable</h3>
124 + </div>
125 + <p class="text-sm text-base-content/80">
126 + The <a href="/failed" class="link">Work</a> page shows live queue
127 + depth, consumer counts, recent completions per kind, and every failed
128 + job. One click retries a job, drains a DLQ back into the main queue,
129 + or replays everything at once.
130 + </p>
131 + </div>
132 +
102 133 <div class="border border-base-300 rounded-box p-5 bg-base-200">
103 134 <div class="flex items-center gap-2 mb-2">
104 135 <i class="fas fa-fw fa-fingerprint size-5 text-accent"></i>
@@ -110,6 +141,18 @@
110 141 bytes, no duplicate categorizer runs.
111 142 </p>
112 143 </div>
144 +
145 + <div class="border border-base-300 rounded-box p-5 bg-base-200">
146 + <div class="flex items-center gap-2 mb-2">
147 + <i class="fas fa-fw fa-code size-5 text-accent"></i>
148 + <h3 class="font-semibold">JSON API parity</h3>
149 + </div>
150 + <p class="text-sm text-base-content/80">
151 + Every HTML surface — uploads, search, tags, retries, DLQ requeue —
152 + has a JSON twin under <code>/api</code>, authenticated with the same
153 + bearer token the Android app uses. Build whatever client you want.
154 + </p>
155 + </div>
113 156 </div>
114 157 </section>
115 158
@@ -136,9 +179,10 @@
136 179 <span class="font-mono text-xs text-accent">02</span>
137 180 <h3 class="font-semibold">Drop something in</h3>
138 181 <p class="text-sm text-base-content/80">
139 − Visit <a href="/uploads" class="link">Uploads</a>, pick a file, add
140 − optional tags. Thumbnails, OCR, and describe jobs fire off in the
141 − background — watch the status grid on the detail page.
182 + Upload from <a href="/uploads" class="link">the web</a>, or pair the
183 + Android app from <a href="/api-tokens" class="link">/api-tokens</a>
184 + and share images straight from your gallery. Categorizer jobs fire off
185 + in the background — watch the status grid on the detail page.
142 186 </p>
143 187 </li>
144 188 <li class="border border-base-300 rounded-box p-5 bg-base-200 flex flex-col gap-2">
@@ -158,8 +202,9 @@
158 202 <p class="text-base-content/70 max-w-2xl mx-auto">
159 203 axum + sea-orm on Postgres, RabbitMQ for the categorizer queues,
160 204 <a href="https://github.com/robertknight/ocrs" class="link">ocrs</a>
161 − for OCR, minijinja + Tailwind + DaisyUI on the front. Three binaries:
162 − webservice, queue-worker, registration-token.
205 + for OCR, minijinja + Tailwind + DaisyUI on the front, a Kotlin/Compose
206 + Android client. Three binaries: webservice, queue-worker,
207 + registration-token.
163 208 </p>
164 209 </section>
165 210 </div>
modified templates/failed.html
+25 −1
@@ -5,8 +5,18 @@
5 5
6 6 {# Queue depth — single line per kind, "messages / consumers, dlq=N". #}
7 7 {% if depths %}
8 + {% set total_dlq = depths|map(attribute="dlq_messages")|sum %}
8 9 <div>
9 − <h3 class="text-lg font-bold text-accent mb-2">Queue depth</h3>
10 + <div class="flex items-center justify-between mb-2">
11 + <h3 class="text-lg font-bold text-accent">Queue depth</h3>
12 + {% if total_dlq > 0 %}
13 + <form method="post" action="/failed/dlq/requeue" class="m-0">
14 + <button type="submit" class="btn btn-warning btn-xs">
15 + <i class="fas fa-rotate-left"></i> Requeue all DLQs ({{ total_dlq }})
16 + </button>
17 + </form>
18 + {% endif %}
19 + </div>
10 20 <div class="overflow-x-auto">
11 21 <table class="table table-sm">
12 22 <thead>
@@ -15,6 +25,7 @@
15 25 <th class="text-right">Pending</th>
16 26 <th class="text-right">Consumers</th>
17 27 <th class="text-right">DLQ</th>
28 + <th class="text-right">Actions</th>
18 29 </tr>
19 30 </thead>
20 31 <tbody>
@@ -24,6 +35,17 @@
24 35 <td class="text-right font-mono">{{ q.messages }}</td>
25 36 <td class="text-right font-mono">{{ q.consumers }}</td>
26 37 <td class="text-right font-mono {% if q.dlq_messages > 0 %}text-error{% endif %}">{{ q.dlq_messages }}</td>
38 + <td class="text-right">
39 + {% if q.dlq_messages > 0 %}
40 + <form method="post" action="/failed/dlq/{{ q.kind }}/requeue" class="inline m-0">
41 + <button type="submit" class="btn btn-ghost btn-xs">
42 + <i class="fas fa-rotate-left"></i> Requeue
43 + </button>
44 + </form>
45 + {% else %}
46 + <span class="opacity-30">—</span>
47 + {% endif %}
48 + </td>
27 49 </tr>
28 50 {% endfor %}
29 51 </tbody>
@@ -65,6 +87,8 @@
65 87
66 88 <div class="flex justify-between items-center mb-4">
67 89 <h2 class="text-2xl font-bold text-accent">Failed jobs ({{ rows|length }})</h2>
90 + {# DB-tracked failed jobs (per-user) — distinct from the DLQ table above,
91 + which holds broker-level dead-lettered messages across all users. #}
68 92 {% if rows %}
69 93 <form method="post" action="/failed/retry" class="m-0">
70 94 <button type="submit" class="btn btn-warning btn-sm">

Parents: 758983a