Enhance: list every photo's open jobs (jobs manager) #211

Merged
cameron merged 1 commits from feat/enhance-jobs-list into master 2026-11-02 13:28:52 +00:00
5 changed files with 156 additions and 8 deletions

No files matched your search

+4
View File
@@ -516,6 +516,10 @@ GET /enhance/templates
POST /enhance {"path","template","prompt"?,"reference_path"?,"library"?} -> job POST /enhance {"path","template","prompt"?,"reference_path"?,"library"?} -> job
GET /enhance/jobs/{id} -> job GET /enhance/jobs/{id} -> job
GET /enhance/jobs?path=&library= -> {"jobs": [job, ...]} newest first, discarded hidden GET /enhance/jobs?path=&library= -> {"jobs": [job, ...]} newest first, discarded hidden
GET /enhance/jobs[?status=a,b] -> {"jobs": [...]} no path = the jobs manager: every photo's
queued/running/unavailable jobs + ready/failed/expired ones touched in the last 3 days
(OPEN_JOBS_WINDOW_SECS), max 100, newest first, discarded + approved hidden; DB only.
`status` filters either form.
GET /enhance/prompts?templates=a,b&limit=10 (templates optional = all; limit 1..50, default 10) GET /enhance/prompts?templates=a,b&limit=10 (templates optional = all; limit 1..50, default 10)
-> {"prompts": [{"prompt","template","last_used","uses"}]} distinct trimmed prompts, newest -> {"prompts": [{"prompt","template","last_used","uses"}]} distinct trimmed prompts, newest
use first; template/last_used from the newest job; discarded + failed jobs count; use first; template/last_used from the newest job; discarded + failed jobs count;
+50
View File
@@ -121,6 +121,11 @@ pub trait EnhanceJobDao: Sync + Send {
) -> Result<Vec<EnhanceJobRow>, DbError>; ) -> Result<Vec<EnhanceJobRow>, DbError>;
/// Jobs the poller still has to advance (`queued` / `running`). /// Jobs the poller still has to advance (`queued` / `running`).
fn list_active(&mut self) -> Result<Vec<EnhanceJobRow>, DbError>; fn list_active(&mut self) -> Result<Vec<EnhanceJobRow>, DbError>;
/// The jobs manager's list, across photos, newest first, discarded hidden:
/// every queued / running / unavailable job, plus ready / failed / expired
/// ones last touched at or after `since` (unix seconds). Approved jobs are
/// done and never listed.
fn list_open(&mut self, since: i64, limit: i64) -> Result<Vec<EnhanceJobRow>, DbError>;
fn update(&mut self, id: &str, update: &JobUpdate) -> Result<(), DbError>; fn update(&mut self, id: &str, update: &JobUpdate) -> Result<(), DbError>;
/// Distinct non-empty prompts (compared trimmed), newest use first. /// Distinct non-empty prompts (compared trimmed), newest use first.
/// Counts every job, discarded and failed included. `templates` limits /// Counts every job, discarded and failed included. `templates` limits
@@ -237,6 +242,23 @@ impl EnhanceJobDao for SqliteEnhanceJobDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e)) .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
} }
fn list_open(&mut self, since: i64, limit: i64) -> Result<Vec<EnhanceJobRow>, DbError> {
trace_db_call("query", "enhance_job_list_open", |_span| {
let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao");
sql_query(format!(
"SELECT {COLUMNS} FROM enhance_jobs WHERE discarded = 0 AND ( \
status IN ('queued', 'running', 'unavailable') \
OR (status IN ('ready', 'failed', 'expired') AND updated_at >= ?)) \
ORDER BY created_at DESC, rowid DESC LIMIT ?"
))
.bind::<BigInt, _>(since)
.bind::<BigInt, _>(limit)
.load::<EnhanceJobRow>(&mut *conn)
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn update(&mut self, id: &str, u: &JobUpdate) -> Result<(), DbError> { fn update(&mut self, id: &str, u: &JobUpdate) -> Result<(), DbError> {
trace_db_call("update", "enhance_job_update", |_span| { trace_db_call("update", "enhance_job_update", |_span| {
let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao"); let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao");
@@ -412,6 +434,34 @@ mod tests {
assert!(d.get("b").unwrap().is_none()); assert!(d.get("b").unwrap().is_none());
} }
#[test]
fn list_open_spans_photos_and_drops_done_old_and_discarded() {
let mut d = dao();
let mut old_ready = job("old", "h1", "ready", 10);
old_ready.updated_at = 10;
d.insert(&old_ready).unwrap();
let mut old_queued = job("stuck", "h2", "queued", 20);
old_queued.updated_at = 20;
d.insert(&old_queued).unwrap();
d.insert(&job("ready", "h1", "ready", 200)).unwrap();
d.insert(&job("failed", "h2", "failed", 300)).unwrap();
d.insert(&job("done", "h3", "approved", 400)).unwrap();
d.insert(&job("gone", "h3", "ready", 500)).unwrap();
d.discard("gone").unwrap();
d.insert(&job("down", "h4", "unavailable", 600)).unwrap();
let ids =
|rows: Vec<EnhanceJobRow>| -> Vec<String> { rows.into_iter().map(|j| j.id).collect() };
// discard() bumps updated_at to now, so `since` well below now still
// can't bring it back.
assert_eq!(
ids(d.list_open(100, 50).unwrap()),
["down", "failed", "ready", "stuck"],
"newest first; active always, finished only since the cutoff"
);
assert_eq!(ids(d.list_open(100, 2).unwrap()), ["down", "failed"]);
}
fn prompt_job(id: &str, template: &str, prompt: Option<&str>, created: i64) -> EnhanceJobRow { fn prompt_job(id: &str, template: &str, prompt: Option<&str>, created: i64) -> EnhanceJobRow {
EnhanceJobRow { EnhanceJobRow {
template: template.into(), template: template.into(),
+36 -8
View File
@@ -347,12 +347,20 @@ pub async fn get_job(
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
pub struct JobsQuery { pub struct JobsQuery {
pub path: String, /// A photo (original or version): that photo's jobs. Omitted: every
/// photo's open jobs, for the jobs manager.
pub path: Option<String>,
pub library: Option<String>, pub library: Option<String>,
/// Comma-separated statuses to keep (e.g. `queued,running,ready`);
/// omitted = all. Applies to both forms.
pub status: Option<String>,
} }
/// A photo's recent jobs, newest first, so the viewer can pick up an enhance /// With `path`: a photo's recent jobs, newest first, so the viewer can pick up
/// in progress after reopening. Discarded jobs are not listed. /// an enhance in progress after reopening. Without: the jobs manager's list,
/// every photo's queued / running / unavailable jobs plus ready / failed /
/// expired ones from the last few days. Discarded and approved-and-done jobs
/// are not listed. DB only.
#[get("/enhance/jobs")] #[get("/enhance/jobs")]
pub async fn list_jobs( pub async fn list_jobs(
_: Claims, _: Claims,
@@ -361,13 +369,33 @@ pub async fn list_jobs(
app_state: web::Data<AppState>, app_state: web::Data<AppState>,
exif_dao: ExifData, exif_dao: ExifData,
) -> impl Responder { ) -> impl Responder {
let file = match resolve_file(&app_state, &exif_dao, &q.path, q.library.as_deref()) { let jobs = match q.path.as_deref() {
Ok(f) => f, Some(path) => {
Err(r) => return r, let file = match resolve_file(&app_state, &exif_dao, path, q.library.as_deref()) {
Ok(f) => f,
Err(r) => return r,
};
svc.jobs_for_hash(&file.hash)
}
None => svc.open_jobs(),
}; };
match svc.jobs_for_hash(&file.hash) { let wanted: Option<Vec<&str>> = q.status.as_deref().map(|s| {
s.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.collect()
});
match jobs {
Ok(jobs) => { Ok(jobs) => {
let views: Vec<JobView> = jobs.iter().map(job_view).collect(); let views: Vec<JobView> = jobs
.iter()
.filter(|j| {
wanted
.as_ref()
.is_none_or(|w| w.contains(&j.status.as_str()))
})
.map(job_view)
.collect();
HttpResponse::Ok().json(json!({ "jobs": views })) HttpResponse::Ok().json(json!({ "jobs": views }))
} }
Err(e) => error_response(&e), Err(e) => error_response(&e),
+50
View File
@@ -914,6 +914,56 @@ async fn discard_hides_the_job_from_the_photo_list_only() {
assert_eq!(st, 404); assert_eq!(st, 404);
} }
#[actix_rt::test]
async fn jobs_without_a_path_list_every_photos_open_jobs() {
let fx = fx().await;
const OTHER: &str = "2009_200/other.JPG";
fx.put(PHOTO, b"photo");
fx.put(OTHER, b"other photo");
let submit = |path: &'static str| {
let fx = &fx;
async move {
let (_, j, _) = call(
fx,
"POST",
"/enhance",
Some(json!({"path": path, "template": "restore"})),
)
.await;
j["id"].as_str().unwrap().to_string()
}
};
let a = submit(PHOTO).await;
let b = submit(OTHER).await;
let c = submit(OTHER).await;
call(&fx, "POST", &format!("/enhance/jobs/{a}/discard"), None).await;
let ids = |v: &Value| -> Vec<String> {
v["jobs"]
.as_array()
.unwrap()
.iter()
.map(|j| j["id"].as_str().unwrap().to_string())
.collect()
};
let (st, all, _) = call(&fx, "GET", "/enhance/jobs", None).await;
assert_eq!(st, 200);
assert_eq!(
ids(&all),
[c.clone(), b.clone()],
"across photos, newest first, discarded hidden"
);
assert_eq!(all["jobs"][0]["path"], OTHER, "each job names its photo");
let (_, none, _) = call(&fx, "GET", "/enhance/jobs?status=ready,failed", None).await;
assert!(
ids(&none).is_empty(),
"status filter keeps only those statuses"
);
let (_, queued, _) = call(&fx, "GET", "/enhance/jobs?status=queued,%20running", None).await;
assert_eq!(ids(&queued).len(), 2);
}
#[actix_rt::test] #[actix_rt::test]
async fn unreachable_active_jobs_become_unavailable_after_the_grace_period() { async fn unreachable_active_jobs_become_unavailable_after_the_grace_period() {
// Submit against the mock, then point nothing at it: use a service whose // Submit against the mock, then point nothing at it: use a service whose
+16
View File
@@ -31,6 +31,11 @@ const IDLE_CHECK_EVERY: Duration = Duration::from_secs(30);
const JOB_RETENTION_SECS: i64 = 14 * 24 * 3600; const JOB_RETENTION_SECS: i64 = 14 * 24 * 3600;
/// Jobs shown per photo. /// Jobs shown per photo.
pub const JOBS_PER_PHOTO: i64 = 20; pub const JOBS_PER_PHOTO: i64 = 20;
/// The jobs manager lists finished-but-unhandled jobs (ready, failed,
/// expired) for this long; active ones always.
pub const OPEN_JOBS_WINDOW_SECS: i64 = 3 * 24 * 3600;
/// Most jobs the jobs manager lists.
pub const OPEN_JOBS_LIMIT: i64 = 100;
pub const MSG_UNAVAILABLE: &str = "ComfyUI isn't running"; pub const MSG_UNAVAILABLE: &str = "ComfyUI isn't running";
pub const MSG_EXPIRED: &str = "This result expired (ComfyUI was restarted). Run it again."; pub const MSG_EXPIRED: &str = "This result expired (ComfyUI was restarted). Run it again.";
@@ -390,6 +395,17 @@ impl EnhanceService {
.map_err(|e| EnhanceError::Internal(e.to_string())) .map_err(|e| EnhanceError::Internal(e.to_string()))
} }
/// Every photo's open jobs for the jobs manager, newest first. DB only:
/// the poller keeps the rows current, so this never calls ComfyUI.
pub fn open_jobs(&self) -> Result<Vec<EnhanceJobRow>, EnhanceError> {
let since = chrono::Utc::now().timestamp() - OPEN_JOBS_WINDOW_SECS;
self.dao
.lock()
.unwrap()
.list_open(since, OPEN_JOBS_LIMIT)
.map_err(|e| EnhanceError::Internal(e.to_string()))
}
/// Distinct recent prompts; DB only, so it works with ComfyUI down. /// Distinct recent prompts; DB only, so it works with ComfyUI down.
pub fn recent_prompts( pub fn recent_prompts(
&self, &self,