neiam /dms

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

q depth

7cd6e39 · gmorell · 2026-05-09 03:01

7 files +215 -4

Files changed

modified src/handlers/failed.rs

+97 −4

@@ -20,7 +20,7 @@ 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, JobKind, JobMessage};
	23	+use crate::queue::{publish_job, queue_depths, JobKind, JobMessage, QueueDepth};
24	24	use crate::AppState;
25	25
26	26	#[derive(Debug, Serialize, Clone)]

@@ -67,14 +67,95 @@ async fn load_failed(

67	67	Ok(out)
68	68	}
69	69
	70	+#[derive(Debug, Serialize, Clone)]
	71	+pub struct CompletedRow {
	72	+ pub upload_id: Uuid,
	73	+ pub kind: String,
	74	+ pub completed_at: Option<chrono::DateTime<chrono::FixedOffset>>,
	75	+ pub file_name_original: String,
	76	+ pub item_path: String,
	77	+}
	78	+
	79	+/// Last N completed (terminal-success) jobs per kind, scoped to the user.
	80	+/// Pulled per kind so a single noisy queue doesn't crowd the others out;
	81	+/// 5 kinds × `per_kind` rows total.
	82	+async fn load_recent_completed(
	83	+ db: &DatabaseConnection,
	84	+ user_id: Uuid,
	85	+ per_kind: u64,
	86	+) -> Result<std::collections::BTreeMap<String, Vec<CompletedRow>>, DbErr> {
	87	+ let mut by_kind = std::collections::BTreeMap::new();
	88	+ for kind in [
	89	+ JobKind::Thumbnail,
	90	+ JobKind::Ocr,
	91	+ JobKind::Describe,
	92	+ JobKind::ExtractTags,
	93	+ JobKind::SimilaritySearch,
	94	+ ] {
	95	+ let kind_str = kind.as_str().to_string();
	96	+ // Pull a generous slice and filter by upload-owner in Rust — joining
	97	+ // through find_also_related is the same shape as load_failed and
	98	+ // keeps the post-filter step uniform across the page.
	99	+ let rows = upload_jobs::Entity::find()
	100	+ .filter(upload_jobs::Column::Status.eq(jobs::STATUS_COMPLETE))
	101	+ .filter(upload_jobs::Column::Kind.eq(kind_str.clone()))
	102	+ .find_also_related(uploads::Entity)
	103	+ .order_by_desc(upload_jobs::Column::CompletedAt)
	104	+ .limit(per_kind * 5) // headroom for non-owner rows
	105	+ .all(db)
	106	+ .await?;
	107	+ let mut taken: Vec<CompletedRow> = Vec::new();
	108	+ for (job, upload) in rows {
	109	+ let Some(upload) = upload else { continue };
	110	+ if upload.user_id != user_id {
	111	+ continue;
	112	+ }
	113	+ taken.push(CompletedRow {
	114	+ upload_id: job.upload_id,
	115	+ kind: job.kind,
	116	+ completed_at: job.completed_at,
	117	+ file_name_original: upload.file_name_original,
	118	+ item_path: upload.item_path,
	119	+ });
	120	+ if taken.len() as u64 >= per_kind {
	121	+ break;
	122	+ }
	123	+ }
	124	+ by_kind.insert(kind_str, taken);
	125	+ }
	126	+ Ok(by_kind)
	127	+}
	128	+
	129	+async fn load_queue_depths(state: &AppState) -> Vec<QueueDepth> {
	130	+ match crate::queue::connect(&state.config.amqp_url).await {
	131	+ Ok(conn) => match queue_depths(&conn).await {
	132	+ Ok(d) => d,
	133	+ Err(e) => {
	134	+ error!(error = %e, "queue_depths failed");
	135	+ Vec::new()
	136	+ }
	137	+ },
	138	+ Err(e) => {
	139	+ error!(error = %e, "amqp connect for queue_depths failed");
	140	+ Vec::new()
	141	+ }
	142	+ }
	143	+}
	144	+
70	145	pub async fn show_failed(
71	146	State(state): State<AppState>,
72	147	user: UserContext,
73	148	) -> Result<Html<String>, StatusCode> {
74		− let rows = load_failed(state.db.as_ref(), user.user.id).await.map_err(\|e\| {
	149	+ let user_id = user.user.id;
	150	+ let rows = load_failed(state.db.as_ref(), user_id).await.map_err(\|e\| {
75	151	error!(error = %e, "load failed jobs failed");
76	152	StatusCode::INTERNAL_SERVER_ERROR
77	153	})?;
	154	+ let recent = load_recent_completed(state.db.as_ref(), user_id, 3).await.map_err(\|e\| {
	155	+ error!(error = %e, "load recent completed jobs failed");
	156	+ StatusCode::INTERNAL_SERVER_ERROR
	157	+ })?;
	158	+ let depths = load_queue_depths(&state).await;
78	159	let template = state.env.get_template("failed.html").map_err(\|e\| {
79	160	error!(error = %e, "load failed.html failed");
80	161	StatusCode::INTERNAL_SERVER_ERROR

@@ -84,6 +165,8 @@ pub async fn show_failed(

84	165	title => "Failed jobs",
85	166	user => user.user,
86	167	rows => rows,
	168	+ recent => recent,
	169	+ depths => depths,
87	170	})
88	171	.map_err(\|e\| {
89	172	error!(error = %e, "render failed.html failed");

@@ -96,11 +179,21 @@ pub async fn api_list_failed(

96	179	State(state): State<AppState>,
97	180	user: UserContext,
98	181	) -> Result<Json<serde_json::Value>, StatusCode> {
99		− let rows = load_failed(state.db.as_ref(), user.user.id).await.map_err(\|e\| {
	182	+ let user_id = user.user.id;
	183	+ let rows = load_failed(state.db.as_ref(), user_id).await.map_err(\|e\| {
100	184	error!(error = %e, "load failed jobs failed");
101	185	StatusCode::INTERNAL_SERVER_ERROR
102	186	})?;
103		− Ok(Json(serde_json::json!({ "total": rows.len(), "items": rows })))
	187	+ let recent = load_recent_completed(state.db.as_ref(), user_id, 3).await.map_err(\|e\| {
	188	+ error!(error = %e, "load recent completed jobs failed");
	189	+ StatusCode::INTERNAL_SERVER_ERROR
	190	+ })?;
	191	+ let depths = load_queue_depths(&state).await;
	192	+ Ok(Json(serde_json::json!({
	193	+ "failed": { "total": rows.len(), "items": rows },
	194	+ "recent_completed": recent,
	195	+ "queues": depths,
	196	+ })))
104	197	}
105	198
106	199	/// Enqueue retries for every failed job belonging to the user. Returns the

modified src/queue/mod.rs

+57 −0

@@ -34,3 +34,60 @@ pub fn queue_name(kind: JobKind) -> String {

34	34	pub fn dlq_name(kind: JobKind) -> String {
35	35	format!("dms.{}.dlq", routing_key(kind).replace('.', "_"))
36	36	}
	37	+
	38	+/// Snapshot of a single RabbitMQ queue. Pulled via passive queue_declare so
	39	+/// the broker reports counts for an existing queue without redeclaring it
	40	+/// (a non-passive declare with mismatched args would error).
	41	+#[derive(Debug, Clone, serde::Serialize)]
	42	+pub struct QueueDepth {
	43	+ pub kind: &'static str,
	44	+ pub queue: String,
	45	+ pub messages: u32,
	46	+ pub consumers: u32,
	47	+ pub dlq: String,
	48	+ pub dlq_messages: u32,
	49	+}
	50	+
	51	+/// Read message + consumer counts for every job-kind queue and its DLQ. A
	52	+/// missing queue (404 from the broker) is reported as zero rather than
	53	+/// failing the whole snapshot — broker may not have seen this kind yet on
	54	+/// a fresh deploy.
	55	+pub async fn queue_depths(conn: &Connection) -> Result<Vec<QueueDepth>, lapin::Error> {
	56	+ use lapin::options::QueueDeclareOptions;
	57	+ use lapin::types::FieldTable;
	58	+
	59	+ let opts = QueueDeclareOptions { passive: true, ..Default::default() };
	60	+ // A passive declare against a non-existent queue raises NOT_FOUND and
	61	+ // closes the channel, so each probe opens a fresh channel. Cheap (no
	62	+ // server round-trip beyond the channel open) and keeps the loop going
	63	+ // when the broker hasn't seen a particular queue yet.
	64	+ async fn probe(conn: &Connection, name: &str, opts: QueueDeclareOptions) -> (u32, u32) {
	65	+ let Ok(ch) = conn.create_channel().await else { return (0, 0); };
	66	+ match ch.queue_declare(name, opts, FieldTable::default()).await {
	67	+ Ok(q) => (q.message_count(), q.consumer_count()),
	68	+ Err(_) => (0, 0),
	69	+ }
	70	+ }
	71	+ let mut out = Vec::new();
	72	+ for kind in [
	73	+ JobKind::Thumbnail,
	74	+ JobKind::Ocr,
	75	+ JobKind::Describe,
	76	+ JobKind::ExtractTags,
	77	+ JobKind::SimilaritySearch,
	78	+ ] {
	79	+ let qname = queue_name(kind);
	80	+ let dlq = dlq_name(kind);
	81	+ let (msgs, consumers) = probe(conn, &qname, opts).await;
	82	+ let (dlq_msgs, _) = probe(conn, &dlq, opts).await;
	83	+ out.push(QueueDepth {
	84	+ kind: kind.as_str(),
	85	+ queue: qname,
	86	+ messages: msgs,
	87	+ consumers,
	88	+ dlq,
	89	+ dlq_messages: dlq_msgs,
	90	+ });
	91	+ }
	92	+ Ok(out)
	93	+}

modified templates/failed.html

+61 −0

@@ -2,6 +2,67 @@

2	2
3	3	{% block inner %}
4	4	<div class="space-y-6 text-left">
	5	+
	6	+ {# Queue depth — single line per kind, "messages / consumers, dlq=N". #}
	7	+ {% if depths %}
	8	+ <div>
	9	+ <h3 class="text-lg font-bold text-accent mb-2">Queue depth</h3>
	10	+ <div class="overflow-x-auto">
	11	+ <table class="table table-sm">
	12	+ <thead>
	13	+ <tr>
	14	+ <th>Kind</th>
	15	+ <th class="text-right">Pending</th>
	16	+ <th class="text-right">Consumers</th>
	17	+ <th class="text-right">DLQ</th>
	18	+ </tr>
	19	+ </thead>
	20	+ <tbody>
	21	+ {% for q in depths %}
	22	+ <tr>
	23	+ <td><span class="badge badge-ghost">{{ q.kind }}</span></td>
	24	+ <td class="text-right font-mono">{{ q.messages }}</td>
	25	+ <td class="text-right font-mono">{{ q.consumers }}</td>
	26	+ <td class="text-right font-mono {% if q.dlq_messages > 0 %}text-error{% endif %}">{{ q.dlq_messages }}</td>
	27	+ </tr>
	28	+ {% endfor %}
	29	+ </tbody>
	30	+ </table>
	31	+ </div>
	32	+ </div>
	33	+ {% endif %}
	34	+
	35	+ {# Recent completions per kind. #}
	36	+ {% if recent %}
	37	+ <div>
	38	+ <h3 class="text-lg font-bold text-accent mb-2">Last 3 completed per queue</h3>
	39	+ <div class="grid grid-cols-1 md:grid-cols-2 gap-3">
	40	+ {% for kind, jobs in recent\|items %}
	41	+ <div class="border border-base-300 rounded-box p-3">
	42	+ <div class="flex items-center justify-between mb-2">
	43	+ <span class="badge badge-ghost">{{ kind }}</span>
	44	+ <span class="text-xs opacity-60">{{ jobs\|length }} shown</span>
	45	+ </div>
	46	+ {% if not jobs %}
	47	+ <div class="text-xs opacity-60">— none yet —</div>
	48	+ {% else %}
	49	+ <ul class="space-y-1">
	50	+ {% for j in jobs %}
	51	+ <li class="text-xs font-mono truncate">
	52	+ <a href="/uploads/{{ j.upload_id }}" class="link link-hover">
	53	+ {{ j.file_name_original }}
	54	+ </a>
	55	+ <span class="opacity-60"> — {{ j.completed_at }}</span>
	56	+ </li>
	57	+ {% endfor %}
	58	+ </ul>
	59	+ {% endif %}
	60	+ </div>
	61	+ {% endfor %}
	62	+ </div>
	63	+ </div>
	64	+ {% endif %}
	65	+
5	66	<div class="flex justify-between items-center mb-4">
6	67	<h2 class="text-2xl font-bold text-accent">Failed jobs ({{ rows\|length }})</h2>
7	68	{% if rows %}

Parents: 94679c7