Enhance: list every photo's open jobs (jobs manager) #211
No files matched your search
@@ -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;
|
||||||
|
|||||||
@@ -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
@@ -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),
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
Reference in new issue
Block a user