From 31bdb7ab97d3b09e143539d55bd9298d49d36665 Mon Sep 17 00:00:00 2001 From: VPRamon Date: Sun, 4 Oct 2026 23:10:01 +0200 Subject: [PATCH 1/2] feat: unify CLI and REST job model Introduce canonical lifecycle, structured job errors, and shared status results for synchronous CLI and asynchronous REST execution.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/workflows/ci.yml | 4 +- docs/architecture/boundaries.md | 2 +- docs/architecture/overview.md | 15 ++ src/bin/spod.rs | 27 ++-- src/service/job.rs | 265 +++++++++++++++++++++++++++++++- src/service/mod.rs | 7 +- src/service/rest.rs | 189 +++++++++++------------ src/service/runner.rs | 20 ++- 8 files changed, 414 insertions(+), 115 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index aec24e2..03f71db 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -148,7 +148,7 @@ jobs: for attempt in {1..60}; do status_json="$(curl --fail --silent "http://127.0.0.1:${port}/jobs/${job_id}")" state="$(printf '%s' "$status_json" | python3 -c 'import json,sys; print(json.load(sys.stdin)["state"])')" - if [[ "$state" == "finished" ]]; then + if [[ "$state" == "succeeded" ]]; then break fi if [[ "$state" == "failed" ]]; then @@ -159,7 +159,7 @@ jobs: done state="$(printf '%s' "$status_json" | python3 -c 'import json,sys; print(json.load(sys.stdin)["state"])')" - [[ "$state" == "finished" ]] + [[ "$state" == "succeeded" ]] run_dir="$SPOD_REST_OUT/$job_id" request_path="$run_dir/request.json" diff --git a/docs/architecture/boundaries.md b/docs/architecture/boundaries.md index 50f9cab..918bac4 100644 --- a/docs/architecture/boundaries.md +++ b/docs/architecture/boundaries.md @@ -4,7 +4,7 @@ |---|---|---| | `siderust::formats` | Standard-format parsing and writing | Service orchestration and transport. | | `siderust::pod` | Force models, propagation, observations, estimation, QC, products, and run metadata. | Service configuration and transport. | -| `spod::service` | Configuration and validation, workflow dispatch, run orchestration, input boundaries, artifact layout, and provenance finalisation using `siderust::pod`. | Reusable numerical algorithms and scientific compatibility facades. | +| `spod::service` | Configuration and validation, canonical job lifecycle/status/error contracts, workflow dispatch, run orchestration, input boundaries, artifact layout, and provenance finalisation using `siderust::pod`. | Reusable numerical algorithms, HTTP semantics, and durable scheduling/persistence. | | `spod` binaries | `clap`-based commands and the REST server; delegates to `spod::service`. | Any numerical logic. | The full set of allowed and forbidden dependency edges is in diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 5becbb0..68e2d6c 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -53,3 +53,18 @@ The CLI and REST binaries are thin front ends over `spod::service::Runner`. They do not assemble pipelines or call scientific components directly. Siderust's scientific APIs remain independently testable, while the service boundary is covered by synthetic end-to-end execution. + +## Shared job model + +`spod::service` owns the canonical `JobId`, `JobState`, `JobStatus`, +`JobResult`, and `JobError` contracts. The lifecycle is `pending -> running -> +succeeded` or `pending -> running -> failed`; terminal statuses carry either +artifact references or a stable machine-readable error category. REST stores +these snapshots in an in-memory `JobStore` and exposes them as JSON. The CLI +uses the same model synchronously and formats its terminal status for +operators. + +The REST adapter is intentionally asynchronous and experimental. It does not +provide durable persistence, scheduling, retries, or distributed workers. +Those concerns are deferred to later service/API work, including the stable +artifact API tracked by #28. diff --git a/src/bin/spod.rs b/src/bin/spod.rs index 74fda8f..e646172 100644 --- a/src/bin/spod.rs +++ b/src/bin/spod.rs @@ -77,17 +77,22 @@ fn main() -> anyhow::Result<()> { } Cmd::Run { config } => { let cfg = spod::service::RunConfig::from_yaml_file(&config)?; - let report = spod::service::run(&cfg, &config)?; - println!( - "OK: ran {} steps, final epoch JD={}, manifest at {}", - report.n_steps, - report - .final_state - .epoch - .to::() - .value(), - report.artifacts.manifest.path.display() - ); + let status = spod::service::execute(&cfg, &config); + match status.state { + spod::service::JobState::Succeeded => { + let result = status.result.expect("succeeded status has a result"); + println!( + "OK: run {} succeeded, manifest at {}", + result.run_id, + result.artifacts.manifest.path.display() + ); + } + spod::service::JobState::Failed => { + let error = status.error.expect("failed status has an error"); + anyhow::bail!("{}: {}", serde_json::to_string(&error.code)?, error.message); + } + state => anyhow::bail!("run ended in unexpected state: {state:?}"), + } } Cmd::InspectManifest { manifest } => { let txt = std::fs::read_to_string(&manifest)?; diff --git a/src/service/job.rs b/src/service/job.rs index ea74281..74c07e5 100644 --- a/src/service/job.rs +++ b/src/service/job.rs @@ -1,15 +1,244 @@ -//! Application-level run requests and results. +//! Application-level run and job contracts. use super::artifacts::Artifacts; use super::config::RunConfig; +use super::error::ServiceError; use super::provenance::RunProvenance; use super::workflow::Workflow; +use serde::{Deserialize, Serialize}; use siderust::astro::dynamics::OrbitState; +use std::collections::HashMap; use std::path::PathBuf; +use std::sync::{Arc, Mutex}; /// Stable identifier assigned by a caller to a run. pub type JobId = String; +/// Stable lifecycle state for a submitted service job. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobState { + /// The service accepted the job but has not started execution. + Pending, + /// The runner is executing the job. + Running, + /// The runner completed successfully. + Succeeded, + /// The runner completed with a categorized failure. + Failed, +} + +/// Stable, machine-readable failure exposed by job APIs. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JobError { + /// Stable category/code for programmatic handling. + pub code: JobErrorCode, + /// Human-readable diagnostic, without implementation-specific details. + pub message: String, +} + +/// Categories of failures that can be returned by a job API. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobErrorCode { + /// Configuration could not be parsed or validated. + Configuration, + /// An input could not be resolved. + Input, + /// The requested workflow or input is unsupported. + Unsupported, + /// An output artifact could not be written or finalized. + Artifact, + /// Scientific execution failed. + Scientific, + /// An unexpected service failure occurred. + Internal, +} + +impl From<&ServiceError> for JobError { + fn from(error: &ServiceError) -> Self { + let code = match error { + ServiceError::Configuration { .. } => JobErrorCode::Configuration, + ServiceError::Unsupported { .. } => JobErrorCode::Unsupported, + ServiceError::Input { .. } => JobErrorCode::Input, + ServiceError::Artifact { .. } => JobErrorCode::Artifact, + ServiceError::Scientific { .. } => JobErrorCode::Scientific, + }; + Self { + code, + message: error.to_string(), + } + } +} + +/// Public result summary attached to a succeeded job. +#[derive(Debug, Clone, Serialize)] +pub struct JobResult { + /// Identifier of the completed run. + pub run_id: JobId, + /// Workflow that produced the result. + pub workflow: Workflow, + /// Generated files and their hashes. + pub artifacts: Artifacts, +} + +impl From<&RunResult> for JobResult { + fn from(result: &RunResult) -> Self { + Self { + run_id: result.run_id.clone(), + workflow: result.workflow, + artifacts: result.artifacts.clone(), + } + } +} + +/// Canonical status snapshot shared by CLI and REST consumers. +#[derive(Debug, Clone, Serialize)] +pub struct JobStatus { + /// Stable service-assigned or caller-provided identifier. + pub id: JobId, + /// Current lifecycle state. + pub state: JobState, + /// Workflow selected for this job. + pub workflow: Workflow, + /// Present after successful execution. + #[serde(skip_serializing_if = "Option::is_none")] + pub result: Option, + /// Present after failed execution. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl JobStatus { + /// Construct a newly accepted job. + pub fn pending(id: JobId, workflow: Workflow) -> Self { + Self { + id, + state: JobState::Pending, + workflow, + result: None, + error: None, + } + } + + /// Advance a pending job to execution. + pub fn start(&mut self) -> Result<(), JobTransitionError> { + if self.state != JobState::Pending { + return Err(JobTransitionError { + from: self.state, + to: JobState::Running, + }); + } + self.state = JobState::Running; + Ok(()) + } + + /// Complete a running job successfully. + pub fn succeed(&mut self, result: JobResult) -> Result<(), JobTransitionError> { + self.require_running(JobState::Succeeded)?; + self.state = JobState::Succeeded; + self.result = Some(result); + Ok(()) + } + + /// Complete a running job with a categorized failure. + pub fn fail(&mut self, error: JobError) -> Result<(), JobTransitionError> { + self.require_running(JobState::Failed)?; + self.state = JobState::Failed; + self.error = Some(error); + Ok(()) + } + + fn require_running(&self, to: JobState) -> Result<(), JobTransitionError> { + if self.state != JobState::Running { + return Err(JobTransitionError { + from: self.state, + to, + }); + } + Ok(()) + } +} + +impl JobStatus { + /// Construct a completed status from a synchronous runner result. + pub fn from_run( + id: JobId, + workflow: Workflow, + result: Result, + ) -> Self { + let mut status = Self::pending(id, workflow); + let _ = status.start(); + match result { + Ok(result) => { + let _ = status.succeed(JobResult::from(&result)); + } + Err(error) => { + let _ = status.fail(JobError::from(&error)); + } + } + status + } +} + +/// An attempted lifecycle transition that is not permitted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct JobTransitionError { + /// State before the attempted transition. + pub from: JobState, + /// Requested destination state. + pub to: JobState, +} + +/// Small synchronized store for the current in-memory job snapshots. +#[derive(Clone, Default)] +pub struct JobStore { + jobs: Arc>>, +} + +impl JobStore { + /// Insert an accepted job. + pub fn insert(&self, status: JobStatus) { + self.with_jobs(|jobs| { + jobs.insert(status.id.clone(), status); + }); + } + + /// Read a status snapshot, if the identifier is known. + pub fn get(&self, id: &str) -> Option { + self.with_jobs(|jobs| jobs.get(id).cloned()) + } + + /// Apply one explicit transition while holding the store lock. + pub fn transition( + &self, + id: &str, + transition: impl FnOnce(&mut JobStatus) -> Result<(), JobTransitionError>, + ) -> Result<(), JobStoreError> { + self.with_jobs(|jobs| { + let status = jobs.get_mut(id).ok_or(JobStoreError::UnknownJob)?; + transition(status).map_err(JobStoreError::InvalidTransition) + }) + } + + fn with_jobs(&self, operation: impl FnOnce(&mut HashMap) -> T) -> T { + let mut jobs = self + .jobs + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + operation(&mut jobs) + } +} + +/// Failure while updating the job store. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum JobStoreError { + /// No job exists for the requested identifier. + UnknownJob, + /// The requested state transition is invalid. + InvalidTransition(JobTransitionError), +} + /// Inputs to the high-level service runner. #[derive(Debug, Clone)] pub struct RunRequest { @@ -42,3 +271,37 @@ pub struct RunResult { /// Application result name retained for callers of the previous service API. pub type RunReport = RunResult; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lifecycle_allows_success_only_from_running() { + let mut status = JobStatus::pending("job-1".into(), Workflow::Synthetic); + assert!(status.start().is_ok()); + assert_eq!(status.state, JobState::Running); + } + + #[test] + fn lifecycle_rejects_invalid_transitions() { + let mut status = JobStatus::pending("job-1".into(), Workflow::Synthetic); + assert!(status.start().is_ok()); + assert!(status.start().is_err()); + assert!(status + .fail(JobError { + code: JobErrorCode::Scientific, + message: "failed".into(), + }) + .is_ok()); + assert_eq!(status.state, JobState::Failed); + assert!(status.start().is_err()); + } + + #[test] + fn service_errors_map_to_stable_codes() { + let error = JobError::from(&ServiceError::unsupported("not available")); + assert_eq!(error.code, JobErrorCode::Unsupported); + assert_eq!(serde_json::to_value(error).unwrap()["code"], "unsupported"); + } +} diff --git a/src/service/mod.rs b/src/service/mod.rs index b2cb74f..f0247bd 100644 --- a/src/service/mod.rs +++ b/src/service/mod.rs @@ -49,7 +49,10 @@ pub mod workflow; pub use artifacts::Artifacts; pub use config::{ForcesConfig, InputsConfig, RunConfig}; pub use error::ServiceError; -pub use job::{JobId, RunReport, RunRequest, RunResult}; +pub use job::{ + JobError, JobErrorCode, JobId, JobResult, JobState, JobStatus, JobStore, JobStoreError, + JobTransitionError, RunReport, RunRequest, RunResult, +}; pub use provenance::RunProvenance; -pub use runner::{run, Runner}; +pub use runner::{execute, run, Runner}; pub use workflow::Workflow; diff --git a/src/service/rest.rs b/src/service/rest.rs index 739a227..9c60194 100644 --- a/src/service/rest.rs +++ b/src/service/rest.rs @@ -1,40 +1,22 @@ -//! # REST interface for POD jobs +//! # REST adapter for POD jobs //! -//! ## Scientific scope +//! REST is an adapter over the service-owned [`JobStatus`] model. It owns +//! request parsing, HTTP status mapping, and the temporary in-memory store, +//! but not lifecycle or scientific execution semantics. //! -//! This module exposes a minimal HTTP surface for launching and querying -//! POD jobs. It does not change the scientific content of the underlying -//! pipeline; it only presents that pipeline through a network-facing -//! control layer suited to local automation and integration tests. -//! -//! The current regime is intentionally narrow: in-memory job state, -//! synthetic job execution, and JSON status payloads. It is not a -//! distributed scheduler or a general production API. -//! -//! ## Technical scope -//! -//! The public items are `JobStatus`, `AppState`, and `router`. The router -//! exposes health and job-submission endpoints, while the worker path -//! delegates actual arc generation and estimation to `spod::service`. -//! -//! Authentication, persistent storage, and arbitrary user-supplied datasets -//! are out of scope. -//! -//! ## References -//! -//! - Fielding, R., Nottingham, M., & Reschke, J. (2022). HTTP Semantics. -//! RFC 9110. -//! - Bray, T. (2017). The JavaScript Object Notation (JSON) Data -//! Interchange Format. RFC 8259. +//! The API is intentionally local and experimental: persistence, +//! authentication, and scheduling are deferred to later issues. #![forbid(unsafe_code)] #![warn(missing_docs)] -use std::collections::HashMap; use std::fs; use std::path::{Path as FsPath, PathBuf}; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; -use crate::service::{ForcesConfig, InputsConfig, RunConfig, RunRequest, Runner, Workflow}; +use crate::service::{ + ForcesConfig, InputsConfig, JobError, JobStatus, JobStore, RunConfig, RunRequest, Runner, + Workflow, +}; use axum::extract::{Path, State}; use axum::http::StatusCode; use axum::response::IntoResponse; @@ -44,30 +26,10 @@ use serde::{Deserialize, Serialize}; use siderust::pod::run::dataset::DatasetRef; use uuid::Uuid; -/// Job status snapshot. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "state", rename_all = "snake_case")] -pub enum JobStatus { - /// Job has been accepted but not yet started. - Pending, - /// Job is currently running. - Running, - /// Job finished successfully. - Finished { - /// Path to the run manifest. - manifest_path: String, - }, - /// Job failed. - Failed { - /// Error message returned by the worker. - error: String, - }, -} - -/// Shared application state (in-memory job table). +/// Shared application state for the REST adapter. #[derive(Clone)] pub struct AppState { - inner: Arc>>, + store: JobStore, /// Base output directory for produced artefacts. pub output_root: Arc, } @@ -76,7 +38,7 @@ impl AppState { /// Create a new state rooted at the given output directory. pub fn new(output_root: PathBuf) -> Self { Self { - inner: Arc::new(Mutex::new(HashMap::new())), + store: JobStore::default(), output_root: Arc::new(output_root), } } @@ -88,11 +50,6 @@ struct JobRequest { enable_j2: bool, } -#[derive(Debug, Serialize)] -struct JobAccepted { - id: String, -} - async fn healthz() -> Json { Json(serde_json::json!({ "status": "ok", @@ -105,23 +62,22 @@ async fn submit_job( body: Option>, ) -> impl IntoResponse { let id = Uuid::new_v4().to_string(); - let req = body.map(|Json(r)| r).unwrap_or_default(); - { - let mut t = state.inner.lock().unwrap(); - t.insert(id.clone(), JobStatus::Pending); - } - let id_clone = id.clone(); + let req = body.map(|Json(request)| request).unwrap_or_default(); + state + .store + .insert(JobStatus::pending(id.clone(), Workflow::Synthetic)); + let state_clone = state.clone(); - tokio::task::spawn_blocking(move || { - run_job(state_clone, id_clone, req); - }); - (StatusCode::ACCEPTED, Json(JobAccepted { id })) + let id_clone = id.clone(); + tokio::task::spawn_blocking(move || run_job(state_clone, id_clone, req)); + + let status = state.store.get(&id).expect("inserted job must exist"); + (StatusCode::ACCEPTED, Json(status)) } async fn job_status(State(state): State, Path(id): Path) -> impl IntoResponse { - let table = state.inner.lock().unwrap(); - match table.get(&id) { - Some(s) => (StatusCode::OK, Json(s.clone())).into_response(), + match state.store.get(&id) { + Some(status) => (StatusCode::OK, Json(status)).into_response(), None => (StatusCode::NOT_FOUND, "unknown job id").into_response(), } } @@ -134,26 +90,24 @@ fn persist_request(output_dir: &FsPath, req: &JobRequest) -> std::io::Result request_ref, Err(error) => { - fail_job(&state, &id, error); + let job_error = JobError { + code: crate::service::JobErrorCode::Input, + message: format!("input resolution failed: {error}"), + }; + let _ = state.store.transition(&id, |status| status.fail(job_error)); return; } }; @@ -178,23 +132,16 @@ fn run_job(state: AppState, id: String, req: JobRequest) { config, config_path: request_ref.path, }); - let mut t = state.inner.lock().unwrap(); match result { Ok(report) => { - t.insert( - id, - JobStatus::Finished { - manifest_path: report.artifacts.manifest.path.display().to_string(), - }, - ); + let job_result = crate::service::JobResult::from(&report); + let _ = state + .store + .transition(&id, |status| status.succeed(job_result)); } - Err(e) => { - t.insert( - id, - JobStatus::Failed { - error: e.to_string(), - }, - ); + Err(error) => { + let job_error = JobError::from(&error); + let _ = state.store.transition(&id, |status| status.fail(job_error)); } } } @@ -211,7 +158,10 @@ pub fn router(state: AppState) -> Router { #[cfg(test)] mod tests { use super::*; + use axum::body::Body; + use axum::http::Request; use std::time::{SystemTime, UNIX_EPOCH}; + use tower::ServiceExt; #[test] fn persisted_request_is_a_real_hashed_manifest_input() { @@ -271,4 +221,49 @@ mod tests { root.join("job-id") ); } + + #[test] + fn accepted_status_has_stable_shape() { + let status = JobStatus::pending("job-id".into(), Workflow::Synthetic); + let value = serde_json::to_value(status).unwrap(); + assert_eq!(value["state"], "pending"); + assert_eq!(value["workflow"], "synthetic"); + assert!(value.get("result").is_none()); + assert!(value.get("error").is_none()); + } + + #[tokio::test] + async fn unknown_job_id_returns_not_found() { + let app = router(AppState::new(std::env::temp_dir())); + let response = app + .oneshot( + Request::builder() + .uri("/jobs/missing") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + + #[test] + fn execution_failures_are_structured() { + let root = std::env::temp_dir().join(format!("spod-rest-failure-{}", std::process::id())); + fs::write(&root, "not a directory").unwrap(); + let state = AppState::new(root.clone()); + let id = "failure-job".to_owned(); + state + .store + .insert(JobStatus::pending(id.clone(), Workflow::Synthetic)); + run_job(state.clone(), id.clone(), JobRequest::default()); + + let status = state.store.get(&id).unwrap(); + assert_eq!(status.state, crate::service::JobState::Failed); + assert_eq!( + status.error.as_ref().unwrap().code, + crate::service::JobErrorCode::Input + ); + fs::remove_file(root).unwrap(); + } } diff --git a/src/service/runner.rs b/src/service/runner.rs index 3738630..bebfe70 100644 --- a/src/service/runner.rs +++ b/src/service/runner.rs @@ -2,7 +2,7 @@ use super::artifacts::{ArtifactLayout, Artifacts}; use super::error::ServiceError; -use super::job::{RunReport, RunRequest}; +use super::job::{JobStatus, RunReport, RunRequest}; use super::pipeline::run_synth; use super::provenance::RunProvenance; use super::synth::{generate, SyntheticArcConfig}; @@ -15,6 +15,13 @@ use siderust::pod::run::dataset::DatasetRef; pub struct Runner; impl Runner { + /// Execute a request and return the canonical terminal job status. + pub fn execute(&self, request: RunRequest) -> JobStatus { + let id = request.config.run_id.clone(); + let workflow = request.config.workflow; + JobStatus::from_run(id, workflow, self.run(request)) + } + /// Execute a request through validation, dispatch, execution, and packaging. pub fn run(&self, request: RunRequest) -> Result { request.config.validate()?; @@ -100,6 +107,17 @@ pub fn run( }) } +/// Execute a request synchronously using the canonical job model. +pub fn execute( + config: &super::config::RunConfig, + config_path: impl Into, +) -> JobStatus { + Runner.execute(RunRequest { + config: config.clone(), + config_path: config_path.into(), + }) +} + fn map_pipeline_error(source: super::pipeline::PipelineError) -> ServiceError { match source { super::pipeline::PipelineError::Io(source) => ServiceError::Artifact { source }, From 456f38cb389cb426c1bb461c25e6612a3bdf5410 Mon Sep 17 00:00:00 2001 From: VPRamon Date: Sun, 4 Oct 2026 23:24:40 +0200 Subject: [PATCH 2/2] test: harden CLI and REST job flows Make REST submission responses deterministically pending, classify service-owned persistence failures as artifacts, and add endpoint and compiled-CLI integration coverage.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- Cargo.toml | 2 +- src/service/job.rs | 16 ++++- src/service/rest.rs | 152 ++++++++++++++++++++++++++++++++------------ tests/cli.rs | 103 ++++++++++++++++++++++++++++++ 4 files changed, 228 insertions(+), 45 deletions(-) create mode 100644 tests/cli.rs diff --git a/Cargo.toml b/Cargo.toml index 5186df9..9064cb0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,7 +48,7 @@ chrono = "0.4" # ── HTTP / REST ──────────────────────────────────────────────────────────────── axum = "0.7" -tokio = { version = "1.39", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1.39", features = ["macros", "rt-multi-thread", "time"] } tower = "0.5" uuid = { version = "1.10", features = ["v4"] } diff --git a/src/service/job.rs b/src/service/job.rs index 74c07e5..2e41206 100644 --- a/src/service/job.rs +++ b/src/service/job.rs @@ -138,6 +138,7 @@ impl JobStatus { self.require_running(JobState::Succeeded)?; self.state = JobState::Succeeded; self.result = Some(result); + self.error = None; Ok(()) } @@ -145,6 +146,7 @@ impl JobStatus { pub fn fail(&mut self, error: JobError) -> Result<(), JobTransitionError> { self.require_running(JobState::Failed)?; self.state = JobState::Failed; + self.result = None; self.error = Some(error); Ok(()) } @@ -168,13 +170,19 @@ impl JobStatus { result: Result, ) -> Self { let mut status = Self::pending(id, workflow); - let _ = status.start(); + status + .start() + .expect("newly constructed job must transition to running"); match result { Ok(result) => { - let _ = status.succeed(JobResult::from(&result)); + status + .succeed(JobResult::from(&result)) + .expect("running job must transition to succeeded"); } Err(error) => { - let _ = status.fail(JobError::from(&error)); + status + .fail(JobError::from(&error)) + .expect("running job must transition to failed"); } } status @@ -295,6 +303,8 @@ mod tests { }) .is_ok()); assert_eq!(status.state, JobState::Failed); + assert!(status.result.is_none()); + assert!(status.error.is_some()); assert!(status.start().is_err()); } diff --git a/src/service/rest.rs b/src/service/rest.rs index 9c60194..c9acd93 100644 --- a/src/service/rest.rs +++ b/src/service/rest.rs @@ -63,15 +63,13 @@ async fn submit_job( ) -> impl IntoResponse { let id = Uuid::new_v4().to_string(); let req = body.map(|Json(request)| request).unwrap_or_default(); - state - .store - .insert(JobStatus::pending(id.clone(), Workflow::Synthetic)); + let status = JobStatus::pending(id.clone(), Workflow::Synthetic); + state.store.insert(status.clone()); let state_clone = state.clone(); let id_clone = id.clone(); tokio::task::spawn_blocking(move || run_job(state_clone, id_clone, req)); - let status = state.store.get(&id).expect("inserted job must exist"); (StatusCode::ACCEPTED, Json(status)) } @@ -91,11 +89,8 @@ fn persist_request(output_dir: &FsPath, req: &JobRequest) -> std::io::Result request_ref, Err(error) => { - let job_error = JobError { - code: crate::service::JobErrorCode::Input, - message: format!("input resolution failed: {error}"), - }; - let _ = state.store.transition(&id, |status| status.fail(job_error)); + let job_error = JobError::from(&crate::service::ServiceError::artifact(error)); + if let Err(transition_error) = + state.store.transition(&id, |status| status.fail(job_error)) + { + log::error!("could not fail persisted job {id}: {transition_error:?}"); + } return; } }; @@ -135,13 +131,20 @@ fn run_job(state: AppState, id: String, req: JobRequest) { match result { Ok(report) => { let job_result = crate::service::JobResult::from(&report); - let _ = state + if let Err(error) = state .store - .transition(&id, |status| status.succeed(job_result)); + .transition(&id, |status| status.succeed(job_result)) + { + log::error!("could not complete job {id}: {error:?}"); + } } Err(error) => { let job_error = JobError::from(&error); - let _ = state.store.transition(&id, |status| status.fail(job_error)); + if let Err(transition_error) = + state.store.transition(&id, |status| status.fail(job_error)) + { + log::error!("could not fail job {id}: {transition_error:?}"); + } } } } @@ -158,8 +161,9 @@ pub fn router(state: AppState) -> Router { #[cfg(test)] mod tests { use super::*; - use axum::body::Body; + use axum::body::{to_bytes, Body}; use axum::http::Request; + use serde_json::Value; use std::time::{SystemTime, UNIX_EPOCH}; use tower::ServiceExt; @@ -222,14 +226,53 @@ mod tests { ); } - #[test] - fn accepted_status_has_stable_shape() { - let status = JobStatus::pending("job-id".into(), Workflow::Synthetic); - let value = serde_json::to_value(status).unwrap(); - assert_eq!(value["state"], "pending"); - assert_eq!(value["workflow"], "synthetic"); - assert!(value.get("result").is_none()); - assert!(value.get("error").is_none()); + #[tokio::test] + async fn health_endpoint_returns_status_and_version() { + let app = router(AppState::new(std::env::temp_dir())); + let response = app + .oneshot( + Request::builder() + .uri("/healthz") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["status"], "ok"); + assert_eq!(value["version"], env!("CARGO_PKG_VERSION")); + } + + #[tokio::test] + async fn submit_returns_pending_and_status_eventually_succeeds() { + let root = std::env::temp_dir().join(format!("spod-rest-http-{}", std::process::id())); + let app = router(AppState::new(root.clone())); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/jobs") + .header("content-type", "application/json") + .body(Body::from(r#"{"enable_j2":false}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::ACCEPTED); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let accepted: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(accepted["state"], "pending"); + assert!(accepted["result"].is_null()); + assert!(accepted["error"].is_null()); + let id = accepted["id"].as_str().unwrap(); + + let status = poll_status(&app, id).await; + assert_eq!(status["state"], "succeeded"); + assert!(status["result"]["artifacts"]["manifest"].is_object()); + fs::remove_dir_all(root).unwrap(); } #[tokio::test] @@ -247,23 +290,50 @@ mod tests { assert_eq!(response.status(), StatusCode::NOT_FOUND); } - #[test] - fn execution_failures_are_structured() { + #[tokio::test] + async fn persistence_failure_reaches_failed_with_artifact_code() { let root = std::env::temp_dir().join(format!("spod-rest-failure-{}", std::process::id())); fs::write(&root, "not a directory").unwrap(); - let state = AppState::new(root.clone()); - let id = "failure-job".to_owned(); - state - .store - .insert(JobStatus::pending(id.clone(), Workflow::Synthetic)); - run_job(state.clone(), id.clone(), JobRequest::default()); - - let status = state.store.get(&id).unwrap(); - assert_eq!(status.state, crate::service::JobState::Failed); - assert_eq!( - status.error.as_ref().unwrap().code, - crate::service::JobErrorCode::Input - ); + let app = router(AppState::new(root.clone())); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/jobs") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::ACCEPTED); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let accepted: Value = serde_json::from_slice(&body).unwrap(); + let status = poll_status(&app, accepted["id"].as_str().unwrap()).await; + assert_eq!(status["state"], "failed"); + assert_eq!(status["error"]["code"], "artifact"); fs::remove_file(root).unwrap(); } + + async fn poll_status(app: &Router, id: &str) -> Value { + for _ in 0..100 { + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/jobs/{id}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + if matches!(value["state"].as_str(), Some("succeeded" | "failed")) { + return value; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("job did not reach a terminal state"); + } } diff --git a/tests/cli.rs b/tests/cli.rs new file mode 100644 index 0000000..87a7ca4 --- /dev/null +++ b/tests/cli.rs @@ -0,0 +1,103 @@ +#![allow(missing_docs)] + +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn test_root(name: &str) -> PathBuf { + let suffix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let root = std::env::temp_dir().join(format!("spod-cli-{name}-{suffix}")); + fs::create_dir_all(&root).unwrap(); + root +} + +fn write_config(root: &Path, body: &str) -> PathBuf { + let path = root.join("run.yaml"); + fs::write(&path, body).unwrap(); + path +} + +fn valid_config(output_dir: &Path) -> String { + format!( + "schema_version: \"1.0.0\"\nrun_id: cli-integration\nworkflow: synthetic\noutput_dir: {output}\ninputs:\n sp3: null\n rinex_obs: null\n rinex_nav: null\n antex: null\nforces:\n two_body: true\n j2: false\n third_body: false\n", + output = output_dir.display() + ) +} + +fn cli(config: &Path, command: &str) -> std::process::Output { + Command::new(env!("CARGO_BIN_EXE_spod")) + .arg(command) + .arg(config) + .output() + .unwrap() +} + +#[test] +fn validate_config_accepts_valid_configuration() { + let root = test_root("validate-success"); + let config = write_config(&root, &valid_config(&root.join("output"))); + let output = cli(&config, "validate-config"); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(String::from_utf8_lossy(&output.stdout).contains("validates")); + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn validate_config_reports_categorized_failure() { + let root = test_root("validate-failure"); + let config = write_config( + &root, + &valid_config(&root.join("output")).replace("1.0.0", "9.9.9"), + ); + let output = cli(&config, "validate-config"); + + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("invalid configuration")); + assert!(stderr.contains("schema_version")); + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn run_reports_success_and_writes_manifest() { + let root = test_root("run-success"); + let output_dir = root.join("output"); + let config = write_config(&root, &valid_config(&output_dir)); + let output = cli(&config, "run"); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(String::from_utf8_lossy(&output.stdout).contains("succeeded")); + assert!(output_dir.join("run.manifest.json").is_file()); + fs::remove_dir_all(root).unwrap(); +} + +#[test] +fn run_reports_shared_unsupported_error_category() { + let root = test_root("run-failure"); + let config = write_config( + &root, + &valid_config(&root.join("output")).replace("sp3: null", "sp3: unsupported.sp3"), + ); + let output = cli(&config, "run"); + + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("unsupported")); + assert!(stderr.contains("real-data ingestion is not implemented")); + fs::remove_dir_all(root).unwrap(); +}