From d86a4d92a81f6f8a7e9f84ca6a4b76f8b2227513 Mon Sep 17 00:00:00 2001 From: TheGeneralist <180094941+thegeneralist01@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:56:10 +0200 Subject: [PATCH 1/4] feat(capture): capture_jobs schema and DB helpers --- crates/archivr-core/src/database.rs | 120 ++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) diff --git a/crates/archivr-core/src/database.rs b/crates/archivr-core/src/database.rs index 7038272..b23d782 100644 --- a/crates/archivr-core/src/database.rs +++ b/crates/archivr-core/src/database.rs @@ -98,6 +98,17 @@ pub struct ApiTokenRecord { pub last_used_at: Option, } +#[derive(Debug, Clone, serde::Serialize)] +pub struct CaptureJobRecord { + pub job_uid: String, + pub archive_id: String, + pub run_uid: Option, + pub status: String, + pub error_text: Option, + pub created_at: String, + pub updated_at: String, +} + pub fn database_path(archive_path: &Path) -> PathBuf { archive_path.join(DATABASE_FILE_NAME) } @@ -240,6 +251,17 @@ pub fn initialize_schema(conn: &Connection) -> Result<()> { PRIMARY KEY (entry_id, tag_id) ); + CREATE TABLE IF NOT EXISTS capture_jobs ( + id INTEGER PRIMARY KEY, + job_uid TEXT NOT NULL UNIQUE, + archive_id TEXT NOT NULL, + run_uid TEXT, + status TEXT NOT NULL CHECK (status IN ('pending','running','completed','failed')) DEFAULT 'pending', + error_text TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_capture_jobs_status ON capture_jobs(status); CREATE INDEX IF NOT EXISTS idx_archive_run_items_run_id ON archive_run_items(run_id); CREATE INDEX IF NOT EXISTS idx_archived_entries_source_identity_id ON archived_entries(source_identity_id); CREATE INDEX IF NOT EXISTS idx_archived_entries_created_by_user_id ON archived_entries(created_by_user_id); @@ -582,6 +604,71 @@ pub fn ensure_default_user(conn: &Connection) -> Result { Ok(conn.last_insert_rowid()) } +/// Creates a pending capture job. Returns the new `job_uid`. +pub fn create_capture_job(conn: &Connection, archive_id: &str) -> Result { + let job_uid = public_id("job"); + let now = now_timestamp(); + conn.execute( + "INSERT INTO capture_jobs (job_uid, archive_id, run_uid, status, error_text, created_at, updated_at) + VALUES (?1, ?2, NULL, 'pending', NULL, ?3, ?3)", + rusqlite::params![job_uid, archive_id, now], + )?; + Ok(job_uid) +} + +/// Updates the status (and optionally run_uid / error_text) of a capture job. +pub fn update_capture_job_status( + conn: &Connection, + job_uid: &str, + status: &str, + run_uid: Option<&str>, + error_text: Option<&str>, +) -> Result<()> { + let now = now_timestamp(); + conn.execute( + "UPDATE capture_jobs SET status = ?1, run_uid = COALESCE(?2, run_uid), + error_text = ?3, updated_at = ?4 WHERE job_uid = ?5", + rusqlite::params![status, run_uid, error_text, now, job_uid], + )?; + Ok(()) +} + +/// Returns a capture job by uid. +pub fn get_capture_job(conn: &Connection, job_uid: &str) -> Result> { + conn.query_row( + "SELECT job_uid, archive_id, run_uid, status, error_text, created_at, updated_at + FROM capture_jobs WHERE job_uid = ?1", + [job_uid], + |row| { + Ok(CaptureJobRecord { + job_uid: row.get(0)?, + archive_id: row.get(1)?, + run_uid: row.get(2)?, + status: row.get(3)?, + error_text: row.get(4)?, + created_at: row.get(5)?, + updated_at: row.get(6)?, + }) + }, + ) + .optional() + .map_err(Into::into) +} + +/// Marks all 'running' capture jobs as 'failed' with a restart message. +/// Called at server startup to clean up jobs interrupted by a previous shutdown. +pub fn fail_stalled_capture_jobs(conn: &Connection) -> Result { + let now = now_timestamp(); + let n = conn.execute( + "UPDATE capture_jobs SET status = 'failed', + error_text = 'interrupted by server restart', + updated_at = ?1 + WHERE status = 'running'", + [now], + )?; + Ok(n) +} + pub fn create_archive_run( conn: &Connection, created_by_user_id: i64, @@ -1568,4 +1655,37 @@ mod tests { let conn = make_auth_conn(); assert!(get_user_for_token(&conn, "unknown").unwrap().is_none()); } + + #[test] + fn capture_job_create_and_get() { + let conn = conn(); + let job_uid = create_capture_job(&conn, "personal").unwrap(); + let job = get_capture_job(&conn, &job_uid).unwrap().unwrap(); + assert_eq!(job.status, "pending"); + assert_eq!(job.archive_id, "personal"); + assert!(job.run_uid.is_none()); + } + + #[test] + fn capture_job_status_transitions() { + let conn = conn(); + let job_uid = create_capture_job(&conn, "test").unwrap(); + update_capture_job_status(&conn, &job_uid, "running", None, None).unwrap(); + update_capture_job_status(&conn, &job_uid, "completed", Some("run_abc"), None).unwrap(); + let job = get_capture_job(&conn, &job_uid).unwrap().unwrap(); + assert_eq!(job.status, "completed"); + assert_eq!(job.run_uid.as_deref(), Some("run_abc")); + } + + #[test] + fn fail_stalled_jobs_on_restart() { + let conn = conn(); + let uid = create_capture_job(&conn, "test").unwrap(); + update_capture_job_status(&conn, &uid, "running", None, None).unwrap(); + let n = fail_stalled_capture_jobs(&conn).unwrap(); + assert_eq!(n, 1); + let job = get_capture_job(&conn, &uid).unwrap().unwrap(); + assert_eq!(job.status, "failed"); + assert!(job.error_text.as_deref().unwrap().contains("interrupted")); + } } From 27b37c3b6d0b75cd51a61610ec61aaac0f8d0be8 Mon Sep 17 00:00:00 2001 From: TheGeneralist <180094941+thegeneralist01@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:56:38 +0200 Subject: [PATCH 2/4] feat(capture): CaptureJobSummary type and archive::get_capture_job --- crates/archivr-core/src/archive.rs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/crates/archivr-core/src/archive.rs b/crates/archivr-core/src/archive.rs index ae26171..23eeb33 100644 --- a/crates/archivr-core/src/archive.rs +++ b/crates/archivr-core/src/archive.rs @@ -60,6 +60,17 @@ pub struct RunSummary { pub error_summary: Option, } +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +pub struct CaptureJobSummary { + pub job_uid: String, + pub archive_id: String, + pub run_uid: Option, + pub status: String, + pub error_text: Option, + pub created_at: String, + pub updated_at: String, +} + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] pub struct Tag { pub tag_uid: String, @@ -299,6 +310,21 @@ pub fn list_runs(conn: &rusqlite::Connection) -> Result> { Ok(runs) } +pub fn get_capture_job( + conn: &rusqlite::Connection, + job_uid: &str, +) -> Result> { + Ok(database::get_capture_job(conn, job_uid)?.map(|r| CaptureJobSummary { + job_uid: r.job_uid, + archive_id: r.archive_id, + run_uid: r.run_uid, + status: r.status, + error_text: r.error_text, + created_at: r.created_at, + updated_at: r.updated_at, + })) +} + /// Resolves an artifact to its absolute on-disk path under `store_path`. /// /// `artifact.relpath` is a store-relative path (e.g. `raw/a/b/abc.pdf`). From cfd51778b1608a0ce4e388289e7d16ffb726b608 Mon Sep 17 00:00:00 2001 From: TheGeneralist <180094941+thegeneralist01@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:57:44 +0200 Subject: [PATCH 3/4] feat(capture): async capture_handler (spawn_blocking) + job poll endpoint --- crates/archivr-server/src/routes.rs | 94 +++++++++++++++++++++++++++-- 1 file changed, 90 insertions(+), 4 deletions(-) diff --git a/crates/archivr-server/src/routes.rs b/crates/archivr-server/src/routes.rs index ed27f8a..2fe9819 100644 --- a/crates/archivr-server/src/routes.rs +++ b/crates/archivr-server/src/routes.rs @@ -103,6 +103,10 @@ pub fn app(registry: ServerRegistry, auth_db_path: std::path::PathBuf) -> Router ) .route("/api/archives/:archive_id/runs", get(list_runs)) .route("/api/archives/:archive_id/captures", post(capture_handler)) + .route( + "/api/archives/:archive_id/capture_jobs/:job_uid", + get(get_capture_job_handler), + ) .route("/api/archives/:archive_id/tags", get(list_tags).post(create_tag_handler)) .route( "/api/archives/:archive_id/entries/:entry_uid/tags", @@ -366,7 +370,7 @@ async fn capture_handler( auth_user: AuthUser, Path(archive_id): Path, Json(body): Json, -) -> Result, ApiError> { +) -> Result<(StatusCode, Json), ApiError> { auth_user.require_role(ROLE_USER)?; if body.locator.trim().is_empty() { return Err(ApiError::bad_request("locator must not be empty")); @@ -374,9 +378,67 @@ async fn capture_handler( let mounted = mounted_archive(&state, &archive_id)?; let archive_paths = archive::read_archive_paths(&mounted.archive_path) .map_err(ApiError::from)?; - let result = capture::perform_capture(&archive_paths, &body.locator, Some(&archive_id)) - .map_err(ApiError::from)?; - Ok(Json(result)) + + // Create job record in the archive DB. + let conn = database::open_or_initialize(&mounted.archive_path)?; + let job_uid = database::create_capture_job(&conn, &archive_id)?; + drop(conn); + + // Spawn background capture. + let locator = body.locator.trim().to_string(); + let archive_path = mounted.archive_path.clone(); + let job_uid_bg = job_uid.clone(); + let archive_id_bg = archive_id.clone(); + tokio::task::spawn_blocking(move || { + let conn = match database::open_or_initialize(&archive_path) { + Ok(c) => c, + Err(e) => { + eprintln!("warn: capture job {job_uid_bg}: db open failed: {e:#}"); + return; + } + }; + database::update_capture_job_status(&conn, &job_uid_bg, "running", None, None).ok(); + match capture::perform_capture(&archive_paths, &locator, Some(&archive_id_bg)) { + Ok(result) => { + database::update_capture_job_status( + &conn, + &job_uid_bg, + "completed", + Some(&result.run_uid), + None, + ) + .ok(); + } + Err(e) => { + database::update_capture_job_status( + &conn, + &job_uid_bg, + "failed", + None, + Some(&format!("{e:#}")), + ) + .ok(); + } + } + }); + + Ok(( + StatusCode::ACCEPTED, + Json(serde_json::json!({ "job_uid": job_uid, "status": "pending" })), + )) +} + +async fn get_capture_job_handler( + State(state): State, + auth_user: AuthUser, + Path((archive_id, job_uid)): Path<(String, String)>, +) -> Result, ApiError> { + auth_user.require_role(ROLE_USER)?; + let mounted = mounted_archive(&state, &archive_id)?; + let conn = database::open_or_initialize(&mounted.archive_path)?; + archive::get_capture_job(&conn, &job_uid)? + .map(Json) + .ok_or_else(|| ApiError::not_found("capture job not found")) } async fn auth_setup_status( @@ -1421,4 +1483,28 @@ mod tests { assert_eq!(response.status(), StatusCode::UNAUTHORIZED); } + + #[tokio::test] + async fn capture_post_returns_accepted_with_job_uid() { + let dir = tempfile::tempdir().unwrap(); + let (registry, _, auth_path) = make_test_registry(&dir); + let session_cookie = make_test_session(&auth_path); + let response = app(registry, auth_path) + .oneshot( + Request::builder() + .method("POST") + .uri("/api/archives/test/captures") + .header("content-type", "application/json") + .header("cookie", &session_cookie) + .body(Body::from(r#"{"locator":"local:/nonexistent"}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::ACCEPTED); + let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert!(json["job_uid"].as_str().is_some(), "response must have job_uid"); + assert_eq!(json["status"], "pending"); + } } From da2b58ef96fa9457040ae6903ab7df735d4e4c91 Mon Sep 17 00:00:00 2001 From: TheGeneralist <180094941+thegeneralist01@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:57:59 +0200 Subject: [PATCH 4/4] feat(capture): mark stalled running jobs failed on server startup --- crates/archivr-server/src/main.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/crates/archivr-server/src/main.rs b/crates/archivr-server/src/main.rs index b4ab90d..ab7366b 100644 --- a/crates/archivr-server/src/main.rs +++ b/crates/archivr-server/src/main.rs @@ -26,6 +26,17 @@ async fn main() -> Result<()> { let app = routes::app(registry.clone(), auth_db_path.clone()); + // On startup, mark any jobs that were 'running' when the server last stopped as 'failed'. + for archive in ®istry.archives { + if let Ok(conn) = archivr_core::database::open_or_initialize(&archive.archive_path) { + match archivr_core::database::fail_stalled_capture_jobs(&conn) { + Ok(n) if n > 0 => eprintln!("info: marked {n} stalled capture job(s) as failed in '{}'", archive.id), + Err(e) => eprintln!("warn: stalled job cleanup failed for '{}': {e:#}", archive.id), + _ => {} + } + } + } + // Spawn session cleanup: runs at startup and every 24h. let cleanup_auth_path = auth_db_path.clone(); tokio::spawn(async move {