fix(sof): spec-conformance fixes from PR HL7/sql-on-fhir#365 assessment

Four fixes found by auditing the implementation against spec PR #365:

- Unsupported _format → 400 (was 415 on the sof-server body path):
  SofError::UnsupportedContentType now maps to ServerError::BadRequest,
  per the spec's "SHALL be rejected with 400 Bad Request +
  OperationOutcome". The stub tests that entrenched 415 are corrected,
  and a new in-bin test locks the 400 against the production handler.

- _format=json export shards are now downloaded with
  Content-Type: application/json (previously fell through to the
  ndjson branch); test asserts the header.

- Export cancellation race: a job cancelled mid-run could be
  resurrected to Completed/Failed by its background task, making
  post-DELETE polls return 200 instead of the spec-required 404.
  Completion/failure/progress writebacks now go through
  set_status_if_running, which leaves a Cancelled job untouched.
  Deterministic unit test added using a blocking mock SofRunner.

- HFS_EXPORT_PRESIGN_TTL_SECS default raised 3600 → 86400 so
  out-of-the-box S3 deployments honor the spec's >= 24h
  output.location validity (matching the Expires header already
  advertised on the completion poll).

Author: smunini

crates/rest/src/config.rs

@@ -583,7 +583,11 @@ pub struct ServerConfig {
     pub export_s3_region: Option<String>,
 
     /// Pre-signed URL TTL (seconds) for S3 export sink.
-    #[arg(long, env = "HFS_EXPORT_PRESIGN_TTL_SECS", default_value = "3600")]
+    ///
+    /// Defaults to 24 hours: the SQL-on-FHIR spec requires `output.location`
+    /// download URLs to remain valid for at least 24 hours after export
+    /// completion (matching the `Expires` header on the completion poll).
+    #[arg(long, env = "HFS_EXPORT_PRESIGN_TTL_SECS", default_value = "86400")]
     pub export_presign_ttl_secs: u64,
 
     /// Maximum concurrent export jobs.
@@ -682,7 +686,7 @@ impl Default for ServerConfig {
             export_dir: "./exports".to_string(),
             export_s3_bucket: None,
             export_s3_region: None,
-            export_presign_ttl_secs: 3600,
+            export_presign_ttl_secs: 86_400,
             export_max_concurrency: 4,
             export_shard_rows: 500_000,
             export_controller: "memory".to_string(),
@@ -793,7 +797,7 @@ impl ServerConfig {
             export_dir: "./exports".to_string(),
             export_s3_bucket: None,
             export_s3_region: None,
-            export_presign_ttl_secs: 3600,
+            export_presign_ttl_secs: 86_400,
             export_max_concurrency: 4,
             export_shard_rows: 500_000,
             export_controller: "memory".to_string(),

crates/rest/src/export/in_memory.rs

@@ -157,8 +157,9 @@ impl<Sink: ExportSink + 'static> ExportJobController for InMemoryController<Sink
                         shards = completed_files.len(),
                         "export job completed"
                     );
-                    jobs.insert(
-                        jid,
+                    set_status_if_running(
+                        &jobs,
+                        &jid,
                         JobStatus::Completed {
                             files: completed_files,
                             submitted_at,
@@ -170,8 +171,9 @@ impl<Sink: ExportSink + 'static> ExportJobController for InMemoryController<Sink
                 }
                 Err(message) => {
                     warn!(job_id = %jid, error = %message, "export job failed");
-                    jobs.insert(
-                        jid,
+                    set_status_if_running(
+                        &jobs,
+                        &jid,
                         JobStatus::Failed {
                             message,
                             submitted_at,
@@ -236,8 +238,23 @@ fn ext_for(format: &str) -> &'static str {
     }
 }
 
+/// Transitions `jid` to `status` only if the job is still `Running`.
+///
+/// A job cancelled mid-run keeps its `Cancelled` state: the spec requires
+/// status polls after a DELETE to return 404, so a background task that
+/// finishes anyway must not resurrect the job to Completed/Failed.
+fn set_status_if_running(jobs: &DashMap<String, JobStatus>, jid: &str, status: JobStatus) {
+    if let Some(mut entry) = jobs.get_mut(jid) {
+        if matches!(&*entry, JobStatus::Running { .. }) {
+            *entry = status;
+        }
+    }
+}
+
 /// Records job progress, capped at 99 while running so callers don't see
 /// "100%" until the manifest is actually available from the status poll.
+/// A job that is no longer `Running` (e.g. cancelled mid-run) is left
+/// untouched — see [`set_status_if_running`].
 fn record_progress(
     jobs: &DashMap<String, JobStatus>,
     jid: &str,
@@ -246,8 +263,9 @@ fn record_progress(
     total: u32,
 ) {
     let percent = ((done * 100) / total.max(1)).min(99) as u8;
-    jobs.insert(
-        jid.to_string(),
+    set_status_if_running(
+        jobs,
+        jid,
         JobStatus::Running {
             percent,
             submitted_at,
@@ -648,3 +666,86 @@ fn csv_cell(v: &serde_json::Value) -> String {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::export::sink::InMemorySink;
+    use async_trait::async_trait;
+    use helios_persistence::core::sof_runner::{RowStream, SofError, ViewFilters};
+    use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions};
+    use tokio::sync::Notify;
+
+    /// A `SofRunner` that blocks until `release` is notified, then yields an
+    /// empty row stream. Lets a test hold a job in the Running state for as
+    /// long as it needs.
+    struct BlockingRunner {
+        release: Arc<Notify>,
+    }
+
+    #[async_trait]
+    impl SofRunner for BlockingRunner {
+        async fn run_view(
+            &self,
+            _tenant: &TenantContext,
+            _view_definition: serde_json::Value,
+            _filters: ViewFilters,
+        ) -> Result<RowStream, SofError> {
+            self.release.notified().await;
+            Ok(Box::pin(futures::stream::empty()))
+        }
+
+        fn runner_name(&self) -> &'static str {
+            "blocking-test-runner"
+        }
+    }
+
+    /// Spec (#363): status polls after a DELETE return 404, so a job
+    /// cancelled while running must stay Cancelled — the background task
+    /// finishing later must not overwrite the state with Completed.
+    #[tokio::test]
+    async fn cancelled_job_is_not_resurrected_by_late_completion() {
+        let release = Arc::new(Notify::new());
+        let runner = Arc::new(BlockingRunner {
+            release: Arc::clone(&release),
+        });
+        let controller =
+            InMemoryController::new(runner, InMemorySink::new("http://localhost"), None);
+
+        let tenant = TenantContext::new(TenantId::new("t1"), TenantPermissions::full_access());
+        let job_id = controller.submit(ExportTask {
+            work: ExportWork::Views(vec![NamedView {
+                name: "patients".to_string(),
+                view: serde_json::json!({
+                    "resourceType": "ViewDefinition",
+                    "resource": "Patient",
+                    "status": "active",
+                    "select": [{"column": [{"name": "id", "path": "id"}]}]
+                }),
+            }]),
+            tenant,
+            filters: ViewFilters::default(),
+            format: "ndjson".to_string(),
+            header: true,
+            client_tracking_id: None,
+        });
+
+        // The runner is blocked, so the job is still Running — cancel it.
+        assert!(controller.cancel("t1", &job_id));
+        assert!(matches!(
+            controller.get_status("t1", &job_id),
+            Some(JobStatus::Cancelled)
+        ));
+
+        // Unblock the background task and give it time to run to completion.
+        // The Cancelled state must survive.
+        release.notify_one();
+        for _ in 0..20 {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            match controller.get_status("t1", &job_id) {
+                Some(JobStatus::Cancelled) => {}
+                other => panic!("cancelled job must stay Cancelled, got {other:?}"),
+            }
+        }
+    }
+}

crates/rest/src/handlers/sof/export.rs

@@ -1285,6 +1285,9 @@ where
                 "application/vnd.apache.parquet"
             } else if filename.ends_with(".fhir.ndjson") {
                 "application/fhir+ndjson"
+            } else if filename.ends_with(".json") {
+                // `_format=json` shards hold a single JSON array of rows.
+                "application/json"
             } else {
                 "application/x-ndjson"
             };

crates/rest/tests/sof_export.rs

@@ -803,6 +803,14 @@ mod sof_export_tests {
             .add_header(X_TENANT_ID, "test-tenant")
             .await;
         assert_eq!(dl.status_code(), StatusCode::OK);
+        // The shard is a JSON array, so it must be served as
+        // `application/json` — not the ndjson fallthrough.
+        assert_eq!(
+            dl.headers()
+                .get("content-type")
+                .and_then(|v| v.to_str().ok()),
+            Some("application/json")
+        );
         let body = dl.text();
         assert!(
             body.trim_start().starts_with('['),

crates/sof/src/error.rs

@@ -73,9 +73,11 @@ impl std::error::Error for ServerError {}
 impl From<SofError> for ServerError {
     fn from(err: SofError) -> Self {
         match &err {
-            SofError::UnsupportedContentType(_) => {
-                ServerError::UnsupportedMediaType(err.to_string())
-            }
+            // Spec (operations-common, Output Formats): an unsupported
+            // `_format` value SHALL be rejected with 400 Bad Request +
+            // OperationOutcome — 415 is reserved for transport-level
+            // Content-Type/Content-Encoding problems.
+            SofError::UnsupportedContentType(_) => ServerError::BadRequest(err.to_string()),
             SofError::InvalidSource(_)
             | SofError::SourceNotFound(_)
             | SofError::UnsupportedSourceProtocol(_) => ServerError::BadRequest(err.to_string()),

crates/sof/src/server.rs

@@ -520,6 +520,38 @@ mod tests {
         assert_eq!(json["service"], "sof-server");
     }
 
+    // ── Unsupported `_format` ─────────────────────────────────────────────
+
+    /// Spec (operations-common, Output Formats): an unsupported `_format`
+    /// value SHALL be rejected with 400 Bad Request + OperationOutcome —
+    /// for the body parameter as well as the query parameter. (The stub
+    /// suite in `tests/` used to entrench 415 for the body path.)
+    #[tokio::test]
+    async fn test_unsupported_body_format_returns_400() {
+        let server = TestServer::new(create_app()).unwrap();
+
+        let mut body = run_request_body();
+        body["parameter"]
+            .as_array_mut()
+            .unwrap()
+            .push(serde_json::json!({"name": "_format", "valueCode": "text/plain"}));
+
+        let response = server
+            .post("/ViewDefinition/$viewdefinition-run")
+            .json(&body)
+            .await;
+
+        assert_eq!(
+            response.status_code(),
+            StatusCode::BAD_REQUEST,
+            "unsupported body _format must be 400, got {}: {}",
+            response.status_code(),
+            response.text()
+        );
+        let json: serde_json::Value = response.json();
+        assert_eq!(json["resourceType"], "OperationOutcome");
+    }
+
     // ── HTTP compression ──────────────────────────────────────────────────
 
     /// A minimal valid `$viewdefinition-run` Parameters body.

crates/sof/tests/common/mod.rs

@@ -449,17 +449,12 @@ async fn run_view_definition_handler(
     // (overly-strict) production behavior; aligned to the new lenient
     // rule per audit item #14.
 
+    // Per spec (operations-common, Output Formats): an unsupported
+    // `_format` value → 400 Bad Request + OperationOutcome (mirrors the
+    // production mapping of SofError::UnsupportedContentType).
     let content_type = match parse_content_type(accept, format, header_param) {
         Ok(ct) => ct,
-        Err(e) => match e {
-            helios_sof::SofError::UnsupportedContentType(_) => {
-                return error_response(
-                    axum::http::StatusCode::UNSUPPORTED_MEDIA_TYPE,
-                    &e.to_string(),
-                );
-            }
-            _ => return error_response(axum::http::StatusCode::BAD_REQUEST, &e.to_string()),
-        },
+        Err(e) => return error_response(axum::http::StatusCode::BAD_REQUEST, &e.to_string()),
     };
 
     // Create ViewDefinition and Bundle

crates/sof/tests/server_tests.rs

@@ -336,7 +336,8 @@ async fn test_run_view_definition_unsupported_format() {
         .json(&request_body)
         .await;
 
-    assert_eq!(response.status_code(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
+    // Spec: unsupported `_format` → 400 Bad Request + OperationOutcome.
+    assert_eq!(response.status_code(), StatusCode::BAD_REQUEST);
 
     let json: serde_json::Value = response.json();
     assert_eq!(json["resourceType"], "OperationOutcome");

crates/sof/tests/test_format_parameter_body.rs

@@ -353,7 +353,8 @@ async fn test_invalid_format_parameter_in_body() {
         .json(&request_body)
         .await;
 
-    assert_eq!(response.status_code(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
+    // Spec: unsupported `_format` → 400 Bad Request + OperationOutcome.
+    assert_eq!(response.status_code(), StatusCode::BAD_REQUEST);
 
     let json: serde_json::Value = response.json();
     assert_eq!(json["resourceType"], "OperationOutcome");