Merge pull request 'Enhance: list every photo's open jobs (jobs manager)' (#211) from feat/enhance-jobs-list into master
Reviewed-on: #211
This commit was merged in pull request #211.
This commit is contained in:
commit
4d9e62d05d
5 files changed
+154
-6
No files matched your search
@@ -516,6 +516,10 @@ GET /enhance/templates
|
||||
POST /enhance {"path","template","prompt"?,"reference_path"?,"library"?} -> job
|
||||
GET /enhance/jobs/{id} -> job
|
||||
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)
|
||||
-> {"prompts": [{"prompt","template","last_used","uses"}]} distinct trimmed prompts, newest
|
||||
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>;
|
||||
/// Jobs the poller still has to advance (`queued` / `running`).
|
||||
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>;
|
||||
/// Distinct non-empty prompts (compared trimmed), newest use first.
|
||||
/// 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))
|
||||
}
|
||||
|
||||
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> {
|
||||
trace_db_call("update", "enhance_job_update", |_span| {
|
||||
let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao");
|
||||
@@ -412,6 +434,34 @@ mod tests {
|
||||
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 {
|
||||
EnhanceJobRow {
|
||||
template: template.into(),
|
||||
|
||||
+34
-6
@@ -347,12 +347,20 @@ pub async fn get_job(
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
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>,
|
||||
/// 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
|
||||
/// in progress after reopening. Discarded jobs are not listed.
|
||||
/// With `path`: a photo's recent jobs, newest first, so the viewer can pick up
|
||||
/// 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")]
|
||||
pub async fn list_jobs(
|
||||
_: Claims,
|
||||
@@ -361,13 +369,33 @@ pub async fn list_jobs(
|
||||
app_state: web::Data<AppState>,
|
||||
exif_dao: ExifData,
|
||||
) -> impl Responder {
|
||||
let file = match resolve_file(&app_state, &exif_dao, &q.path, q.library.as_deref()) {
|
||||
let jobs = match q.path.as_deref() {
|
||||
Some(path) => {
|
||||
let file = match resolve_file(&app_state, &exif_dao, path, q.library.as_deref()) {
|
||||
Ok(f) => f,
|
||||
Err(r) => return r,
|
||||
};
|
||||
match svc.jobs_for_hash(&file.hash) {
|
||||
svc.jobs_for_hash(&file.hash)
|
||||
}
|
||||
None => svc.open_jobs(),
|
||||
};
|
||||
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) => {
|
||||
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 }))
|
||||
}
|
||||
Err(e) => error_response(&e),
|
||||
|
||||
@@ -914,6 +914,56 @@ async fn discard_hides_the_job_from_the_photo_list_only() {
|
||||
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]
|
||||
async fn unreachable_active_jobs_become_unavailable_after_the_grace_period() {
|
||||
// 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;
|
||||
/// Jobs shown per photo.
|
||||
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_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()))
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub fn recent_prompts(
|
||||
&self,
|
||||
|
||||
Reference in new issue
Block a user