diff --git a/Cargo.lock b/Cargo.lock index 40b91a865..6fb2104e9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3584,6 +3584,7 @@ dependencies = [ "mime", "p256 0.13.2", "p384", + "parking_lot", "rand 0.8.6", "regex", "reqwest", diff --git a/crates/rest/Cargo.toml b/crates/rest/Cargo.toml index 1bdf82695..f8ff40032 100644 --- a/crates/rest/Cargo.toml +++ b/crates/rest/Cargo.toml @@ -133,6 +133,12 @@ helios-observability = { path = "../observability" } # Key generation for the JWE round-trip tests rand = "0.8" +# `SearchProvider::search_param_registry` returns a `parking_lot::RwLock`, and +# the batch unit tests' mock storage has to satisfy that trait since #478 +# widened `process_batch`'s bound. Test-only: the crate's own code reaches the +# registry through `helios-persistence` and never names the lock type. +parking_lot = "0.12" + # Temp files tempfile = "3" diff --git a/crates/rest/src/handlers/batch.rs b/crates/rest/src/handlers/batch.rs index 0d04758ec..c4873ce39 100644 --- a/crates/rest/src/handlers/batch.rs +++ b/crates/rest/src/handlers/batch.rs @@ -17,8 +17,8 @@ use helios_audit::{AuditAction, AuditCorrelation, AuditEventBuilder}; use helios_auth::{FhirOperation, Principal, SmartScopePolicy}; use helios_fhir::FhirVersion; use helios_persistence::core::{ - BundleEntry, BundleEntryResult, BundleMethod, BundleProvider, ResourceStorage, - bundle_if_match_gate, + BundleEntry, BundleEntryResult, BundleMethod, BundleProvider, IncludeProvider, ResourceStorage, + RevincludeProvider, SearchProvider, bundle_if_match_gate, }; use helios_persistence::error::{ResourceError, StorageError, TransactionError}; use serde_json::Value; @@ -60,7 +60,13 @@ pub async fn batch_handler( request: Request, ) -> RestResult where - S: ResourceStorage + BundleProvider + helios_persistence::core::SearchProvider + Send + Sync, + S: ResourceStorage + + SearchProvider + + IncludeProvider + + RevincludeProvider + + BundleProvider + + Send + + Sync, { // Extract the Principal from request extensions (set by auth middleware). // If present, per-entry scope checks will be enforced. @@ -251,7 +257,7 @@ async fn process_batch( principal: Option<&Principal>, ) -> RestResult where - S: ResourceStorage + Send + Sync, + S: ResourceStorage + SearchProvider + IncludeProvider + RevincludeProvider + Send + Sync, { debug!( tenant = %tenant.tenant_id(), @@ -371,7 +377,13 @@ async fn process_transaction( principal: Option<&Principal>, ) -> RestResult where - S: ResourceStorage + BundleProvider + helios_persistence::core::SearchProvider + Send + Sync, + S: ResourceStorage + + SearchProvider + + IncludeProvider + + RevincludeProvider + + BundleProvider + + Send + + Sync, { debug!( tenant = %tenant.tenant_id(), @@ -474,12 +486,43 @@ where } } + // GET search entries (`Patient?name=x`, bare `Patient`) cannot run inside + // the storage transaction; the spec orders GETs after all writes, so they + // execute against the just-committed state instead (#478). Their queries + // are still validated up front, where a malformed search can reject the + // whole bundle before anything executes. + let (search_entries, remaining): (Vec<_>, Vec<_>) = indexed_entries.into_iter().partition( + |(_, entry, _): &(usize, BundleEntry, Option)| { + matches!(entry.method, BundleMethod::Get) + && parse_search_entry_url(&entry.url).is_some() + }, + ); + let mut indexed_entries = remaining; + for (index, entry, _) in &search_entries { + let (search_type, pairs) = + parse_search_entry_url(&entry.url).expect("partitioned on is_some"); + let reg = state.storage().search_param_registry(tenant.context()); + let registry = reg.read(); + crate::extractors::build_search_query_from_pairs(&search_type, &pairs, ®istry).map_err( + |e| RestError::BadRequest { + message: format!( + "Entry {}: invalid search '{}': {}", + index, + entry.url, + e.client_response().2 + ), + }, + )?; + } + // Conditional references (`Type?query`) resolve against the server's // content before anything executes, per the transaction processing rules: // exactly one match rewrites the reference to `Type/id`; zero or several // fail the bundle (#459). They used to be stored verbatim — unsearchable // and unresolvable. References to entries created by this same bundle use - // `fullUrl`s, which the storage layer resolves during processing. + // `fullUrl`s, which the storage layer resolves during processing. Runs on + // the write entries only — GET search entries were partitioned out above, + // and their query strings are searches, not conditional references. resolve_conditional_references(state, &tenant, &mut indexed_entries).await?; // Write-path validation: transactions are atomic, so any invalid write @@ -539,6 +582,35 @@ where } } + // GET searches run against the committed state (see above). A + // failure here cannot roll the transaction back, so it surfaces + // as that entry's own error outcome rather than a misleading + // whole-bundle failure for writes that did commit. + let mut search_results: Vec<(usize, BundleEntry, BundleEntryResult)> = + Vec::with_capacity(search_entries.len()); + for (index, entry, _) in &search_entries { + let (search_type, pairs) = + parse_search_entry_url(&entry.url).expect("partitioned on is_some"); + let result = match crate::handlers::search::execute_search_bundle( + state, + &tenant, + &search_type, + pairs, + false, + ) + .await + { + Ok(bundle) => searchset_result(bundle), + // The second of #481's two code-discarding call sites, and + // the more consequential one: this loop bypasses the + // backend executor, so it is the first *reachable* + // per-entry outcome on the transaction arm. Rendered + // through the funnel like every other entry failure. + Err(e) => entry_failure(e), + }; + search_results.push((*index, entry.clone(), result)); + } + // Reorder results back to original entry order let mut ordered_results: Vec<(usize, &BundleEntry, &BundleEntryResult)> = indexed_entries @@ -546,6 +618,9 @@ where .zip(bundle_result.entries.iter()) .map(|((orig_idx, entry, _), result)| (*orig_idx, entry, result)) .collect(); + for (orig_idx, entry, result) in &search_results { + ordered_results.push((*orig_idx, entry, result)); + } ordered_results.sort_by_key(|(idx, _, _)| *idx); for (orig_idx, entry, result) in &ordered_results { @@ -585,10 +660,15 @@ where // rolled-back/internal transaction error never reaches the client // response, the audit trail, or the entry outcome. The raw detail is // preserved server-side by the `error!` log below. + // The search entries join the rollback fan-out: they were + // partitioned out of `indexed_entries` before execution, but the + // audit trail owes every entry of the failed bundle a record. + // (The status/code threading from the stacked #504 refinement + // lands with that PR; main's message-based result stays here.) let (_, _, rollback_reason) = transaction_error_response_parts(&e); let rollback_result = create_error_result(500, &format!("Transaction rolled back: {rollback_reason}")); - for (orig_idx, entry, _) in &indexed_entries { + for (orig_idx, entry, _) in indexed_entries.iter().chain(&search_entries) { let correlation_details = EntryAuditCorrelation::from_bundle(&correlation, *orig_idx); emit_transaction_entry_audit( @@ -667,7 +747,7 @@ async fn process_batch_entry( principal: Option<&Principal>, ) -> BundleEntryResult where - S: ResourceStorage + Send + Sync, + S: ResourceStorage + SearchProvider + IncludeProvider + RevincludeProvider + Send + Sync, { let request = match entry.get("request") { Some(r) => r, @@ -749,6 +829,28 @@ where match method { BundleMethod::Get => { + // A GET entry is either a search (`Patient?name=x`, bare + // `Patient`) or an instance read (`Patient/123`), per the spec's + // "read or search" wording for bundle GETs (#478). + if let Some((search_type, pairs)) = parse_search_entry_url(url) { + return match crate::handlers::search::execute_search_bundle( + state, + tenant, + &search_type, + pairs, + false, + ) + .await + { + Ok(bundle) => searchset_result(bundle), + // Rendered through the funnel like every other entry + // failure. #481 wrote this as `let (status, _, details) = + // e.client_response()` — the same code-discard #504 + // deleted everywhere else, which would have made a search + // entry the one path still answering `processing`. + Err(e) => entry_failure(e), + }; + } // Read operation match state .storage() @@ -1316,6 +1418,54 @@ fn validation_failure_message(error: &RestError) -> String { format!("Validation failed: {error}") } +/// Interprets a bundle-entry GET url as a type-level search, if it is one. +/// +/// Per the FHIR spec, a GET entry may carry any read OR search URL +/// (`Patient?name=x`, or bare `Patient` for an unfiltered type search). +/// Returns the resource type and the parsed query pairs, or `None` when the +/// url addresses a specific instance (`Patient/123`) and should be a read. +fn parse_search_entry_url(url: &str) -> Option<(String, Vec<(String, String)>)> { + let (path, query) = match url.split_once('?') { + Some((p, q)) => (p, Some(q)), + None => (url, None), + }; + let parts: Vec<&str> = path + .trim_start_matches('/') + .split('/') + .filter(|s| !s.is_empty()) + .collect(); + match parts.as_slice() { + [resource_type] => Some(( + resource_type.to_string(), + crate::extractors::query_pairs::parse_query_pairs(query), + )), + _ => None, + } +} + +/// Builds the entry result embedding a searchset Bundle (bundle GET search). +fn searchset_result(bundle: Value) -> BundleEntryResult { + BundleEntryResult { + status: 200, + location: None, + etag: None, + last_modified: None, + resource: Some(bundle), + outcome: None, + } +} + +/// Renders a failed Bundle entry through the status/details pair the +/// single-resource handlers compute. +/// +/// One seam on purpose: the stacked issue-code refinement (#516) upgrades +/// this to the full `OperationOutcome` mapping; until it lands, entries keep +/// main's message-based outcomes. +fn entry_failure(err: RestError) -> BundleEntryResult { + let (status, _, details) = err.client_response(); + create_error_result(status.as_u16(), &details) +} + fn create_error_result(status: u16, message: &str) -> BundleEntryResult { let outcome = serde_json::json!({ "resourceType": "OperationOutcome", @@ -1784,6 +1934,37 @@ mod tests { use super::*; use std::collections::{HashMap, HashSet}; + /// #478: the read-or-search split on a bundle GET url. Instance addresses + /// stay reads; type-level urls (with or without a query) are searches. + #[test] + fn parse_search_entry_url_splits_reads_from_searches() { + let (rt, pairs) = parse_search_entry_url("Patient?name=x&gender=female").expect("search"); + assert_eq!(rt, "Patient"); + assert_eq!(pairs.len(), 2); + + // Bare type: an unfiltered type search, leading slash tolerated. + let (rt, pairs) = parse_search_entry_url("/Patient").expect("bare type search"); + assert_eq!(rt, "Patient"); + assert!(pairs.is_empty()); + + // Instance reads and deeper paths are not searches. + assert!(parse_search_entry_url("Patient/123").is_none()); + assert!(parse_search_entry_url("Patient/123/_history/1").is_none()); + assert!(parse_search_entry_url("").is_none()); + } + + /// #478: the searchset entry result embeds the bundle at 200 with no + /// location/etag baggage. + #[test] + fn searchset_result_embeds_the_bundle() { + let result = searchset_result(serde_json::json!({ + "resourceType": "Bundle", "type": "searchset" + })); + assert_eq!(result.status, 200); + assert!(result.location.is_none()); + assert_eq!(result.resource.unwrap()["type"], "searchset"); + } + use async_trait::async_trait; use helios_audit::AuditSink; use helios_fhir::FhirVersion; @@ -2287,6 +2468,62 @@ mod tests { } } + // #478's search-entry dispatch widened `process_batch`'s bound to + // `SearchProvider + IncludeProvider + RevincludeProvider`, so this mock has + // to satisfy them. Every method is `unimplemented!()`, which is the same + // lever the write methods above use: no unit test in this module drives a + // search entry, and one that started to would panic loudly rather than + // silently exercising a stub. + #[async_trait] + impl helios_persistence::core::SearchProvider for DelayStorage { + async fn search( + &self, + _tenant: &TenantContext, + _query: &helios_persistence::types::SearchQuery, + ) -> StorageResult { + unimplemented!() + } + + async fn search_count( + &self, + _tenant: &TenantContext, + _query: &helios_persistence::types::SearchQuery, + ) -> StorageResult { + unimplemented!() + } + + fn search_param_registry( + &self, + _tenant: &TenantContext, + ) -> Arc> { + unimplemented!() + } + } + + #[async_trait] + impl helios_persistence::core::IncludeProvider for DelayStorage { + async fn resolve_includes( + &self, + _tenant: &TenantContext, + _resources: &[StoredResource], + _includes: &[helios_persistence::types::IncludeDirective], + ) -> StorageResult> { + unimplemented!() + } + } + + #[async_trait] + impl helios_persistence::core::RevincludeProvider for DelayStorage { + async fn resolve_revincludes( + &self, + _tenant: &TenantContext, + _resources: &[StoredResource], + _revincludes: &[helios_persistence::types::IncludeDirective], + ) -> StorageResult> { + unimplemented!() + } + } + /// A batch Bundle of `count` GET entries, targeting `Patient/p0..p{count}`. fn get_bundle(count: usize) -> Value { let entries: Vec = (0..count) @@ -2309,7 +2546,7 @@ mod tests { principal: Option<&Principal>, ) -> Value where - S: ResourceStorage + Send + Sync, + S: ResourceStorage + SearchProvider + IncludeProvider + RevincludeProvider + Send + Sync, { let tenant = TenantExtractor::new("test-tenant", crate::tenant::TenantSource::Default); let response = process_batch( diff --git a/crates/rest/src/handlers/search.rs b/crates/rest/src/handlers/search.rs index d7adc31e9..a136bb94a 100644 --- a/crates/rest/src/handlers/search.rs +++ b/crates/rest/src/handlers/search.rs @@ -156,6 +156,29 @@ async fn execute_search( format: FhirFormat, strict: bool, ) -> RestResult +where + S: ResourceStorage + SearchProvider + IncludeProvider + RevincludeProvider + Send + Sync, +{ + let bundle_json = execute_search_bundle(state, &tenant, resource_type, pairs, strict).await?; + format_resource_response(StatusCode::OK, HeaderMap::new(), &bundle_json, format).map_err(|_| { + RestError::InternalError { + message: "Failed to serialize response".to_string(), + } + }) +} + +/// Executes a type-level search and returns the searchset Bundle as JSON. +/// +/// The HTTP search handlers wrap this in content negotiation; bundle +/// processing (`GET [type]?params` entries in batch/transaction Bundles, +/// #478) embeds the returned Bundle as an entry resource. +pub(crate) async fn execute_search_bundle( + state: &AppState, + tenant: &TenantExtractor, + resource_type: &str, + pairs: Vec<(String, String)>, + strict: bool, +) -> RestResult where S: ResourceStorage + SearchProvider + IncludeProvider + RevincludeProvider + Send + Sync, { @@ -427,11 +450,7 @@ where append_ignored_params_outcome(&mut bundle_json, &ignored_params); } - format_resource_response(StatusCode::OK, HeaderMap::new(), &bundle_json, format).map_err(|_| { - RestError::InternalError { - message: "Failed to serialize response".to_string(), - } - }) + Ok(bundle_json) } /// Appends a `search.mode = outcome` entry to a searchset bundle reporting the diff --git a/crates/rest/tests/batch_conformance.rs b/crates/rest/tests/batch_conformance.rs index f5568b441..c5841df03 100644 --- a/crates/rest/tests/batch_conformance.rs +++ b/crates/rest/tests/batch_conformance.rs @@ -1353,3 +1353,157 @@ mod entry_methods { transaction.assert_status(StatusCode::METHOD_NOT_ALLOWED); } } + +// ============================================================================= +// GET Search Entry Tests (#478) +// ============================================================================= + +mod search_entries { + use super::*; + + #[tokio::test] + async fn test_batch_get_search_entry_returns_searchset() { + let (server, backend) = create_test_server().await; + seed_patient(&backend, "p1", "Nguyen").await; + seed_patient(&backend, "p2", "Smith").await; + + let bundle = json!({ + "resourceType": "Bundle", + "type": "batch", + "entry": [{ + "request": { "method": "GET", "url": "Patient?family=Nguyen" } + }] + }); + + let body = post_batch(&server, bundle).await; + let entry = &body["entry"][0]; + + assert_eq!(entry["response"]["status"].as_str().unwrap(), "200 OK"); + let searchset = &entry["resource"]; + assert_eq!(searchset["resourceType"].as_str().unwrap(), "Bundle"); + assert_eq!(searchset["type"].as_str().unwrap(), "searchset"); + assert_eq!(searchset["entry"].as_array().unwrap().len(), 1); + assert_eq!( + searchset["entry"][0]["resource"]["name"][0]["family"] + .as_str() + .unwrap(), + "Nguyen" + ); + } + + #[tokio::test] + async fn test_batch_get_bare_type_is_a_search() { + let (server, backend) = create_test_server().await; + seed_patient(&backend, "p1", "Nguyen").await; + seed_patient(&backend, "p2", "Smith").await; + + let bundle = json!({ + "resourceType": "Bundle", + "type": "batch", + "entry": [{ + "request": { "method": "GET", "url": "Patient" } + }] + }); + + let body = post_batch(&server, bundle).await; + let searchset = &body["entry"][0]["resource"]; + + assert_eq!(searchset["type"].as_str().unwrap(), "searchset"); + assert_eq!(searchset["entry"].as_array().unwrap().len(), 2); + } + + #[tokio::test] + async fn test_batch_mixes_search_and_read_entries() { + let (server, backend) = create_test_server().await; + seed_patient(&backend, "p1", "Nguyen").await; + + let bundle = json!({ + "resourceType": "Bundle", + "type": "batch", + "entry": [ + { "request": { "method": "GET", "url": "Patient/p1" } }, + { "request": { "method": "GET", "url": "Patient?family=Nguyen" } } + ] + }); + + let body = post_batch(&server, bundle).await; + + let read = &body["entry"][0]; + assert_eq!(read["response"]["status"].as_str().unwrap(), "200 OK"); + assert_eq!( + read["resource"]["resourceType"].as_str().unwrap(), + "Patient" + ); + + let search = &body["entry"][1]; + assert_eq!(search["response"]["status"].as_str().unwrap(), "200 OK"); + assert_eq!(search["resource"]["type"].as_str().unwrap(), "searchset"); + } + + #[tokio::test] + async fn test_transaction_get_search_sees_the_bundles_own_writes() { + let (server, _backend) = create_test_server().await; + + let bundle = json!({ + "resourceType": "Bundle", + "type": "transaction", + "entry": [ + { + "resource": { + "resourceType": "Patient", + "name": [{"family": "Tran"}] + }, + "request": { "method": "POST", "url": "Patient" } + }, + { "request": { "method": "GET", "url": "Patient?family=Tran" } } + ] + }); + + let body = post_batch(&server, bundle).await; + assert_eq!(body["type"].as_str().unwrap(), "transaction-response"); + + let created = &body["entry"][0]; + assert_eq!( + created["response"]["status"].as_str().unwrap(), + "201 Created" + ); + + let search = &body["entry"][1]; + assert_eq!(search["response"]["status"].as_str().unwrap(), "200 OK"); + let searchset = &search["resource"]; + assert_eq!(searchset["type"].as_str().unwrap(), "searchset"); + assert_eq!( + searchset["entry"].as_array().unwrap().len(), + 1, + "the search runs after the writes and must see the created patient" + ); + assert_eq!( + searchset["entry"][0]["resource"]["name"][0]["family"] + .as_str() + .unwrap(), + "Tran" + ); + } + + #[tokio::test] + async fn test_transaction_get_by_id_still_reads_in_transaction() { + let (server, backend) = create_test_server().await; + seed_patient(&backend, "p1", "Nguyen").await; + + let bundle = json!({ + "resourceType": "Bundle", + "type": "transaction", + "entry": [{ + "request": { "method": "GET", "url": "Patient/p1" } + }] + }); + + let body = post_batch(&server, bundle).await; + let entry = &body["entry"][0]; + assert_eq!(entry["response"]["status"].as_str().unwrap(), "200 OK"); + assert_eq!( + entry["resource"]["resourceType"].as_str().unwrap(), + "Patient" + ); + } +}