diff --git a/Cargo.lock b/Cargo.lock index ed472b8..a65a8a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1698,6 +1698,7 @@ dependencies = [ "mime", "multer", "pin-project", + "reqwest 0.13.3", "sentry", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 6a45930..a7f8689 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,6 +63,7 @@ aws-credential-types = { version = "1.2.13", optional = true } testcontainers = "0.27.3" dicom-test-files = "0.4.0" dicom-web = "0.5.0" +reqwest = { version = "0.13.2", default-features = false, features = ["json"] } [lints.rust] unsafe_code = "forbid" diff --git a/src/api/mod.rs b/src/api/mod.rs index 657f1db..85f4eb3 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -14,6 +14,7 @@ mod aets; mod home; pub mod mwl; pub mod qido; +pub mod stgcmt; pub mod stow; pub mod wado; @@ -27,7 +28,8 @@ pub fn routes(base_path: &str) -> Router { .merge(qido::routes()) .merge(wado::routes()) .merge(stow::routes()) - .merge(mwl::routes()), + .merge(mwl::routes()) + .merge(stgcmt::routes()), ); // axum no longer supports nesting at the root diff --git a/src/api/stgcmt/mod.rs b/src/api/stgcmt/mod.rs new file mode 100644 index 0000000..f87ca45 --- /dev/null +++ b/src/api/stgcmt/mod.rs @@ -0,0 +1,5 @@ +mod routes; +mod service; + +pub use routes::routes; +pub use service::*; diff --git a/src/api/stgcmt/routes.rs b/src/api/stgcmt/routes.rs new file mode 100644 index 0000000..73439b1 --- /dev/null +++ b/src/api/stgcmt/routes.rs @@ -0,0 +1,84 @@ +use crate::api::stgcmt::{CommitRequest, CommitmentState}; +use crate::backend::ServiceProvider; +use crate::utils::dicom_json::DicomJsonBody; +use crate::AppState; +use axum::body::Body; +use axum::extract::Path; +use axum::http::header::CONTENT_TYPE; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use axum::Router; +use dicom::object::InMemDicomObject; +use dicom_json::DicomJson; +use tracing::instrument; + +/// HTTP Router for the Storage Commitment Service. +/// +pub fn routes() -> Router { + Router::new().route( + "/commitment-requests/{transaction_uid}", + post(commit).get(check_result), + ) +} + +/// Commit Transaction. +/// +#[instrument(skip_all)] +async fn commit( + provider: ServiceProvider, + Path((_aet, transaction_uid)): Path<(String, String)>, + DicomJsonBody(object): DicomJsonBody, +) -> Response { + let Some(stgcmt) = provider.stgcmt else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "Storage Commitment endpoint is disabled", + ) + .into_response(); + }; + + let request = match CommitRequest::from_object(transaction_uid, &object) { + Ok(request) => request, + Err(err) => return err.into_response(), + }; + + match stgcmt.commit(request).await { + Ok(()) => StatusCode::ACCEPTED.into_response(), + Err(err) => err.into_response(), + } +} + +/// Check Commit Result Transaction. +/// +#[instrument(skip_all)] +async fn check_result( + provider: ServiceProvider, + Path((_aet, transaction_uid)): Path<(String, String)>, +) -> Response { + let Some(stgcmt) = provider.stgcmt else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "Storage Commitment endpoint is disabled", + ) + .into_response(); + }; + + match stgcmt.check_result(&transaction_uid).await { + None => ( + StatusCode::NOT_FOUND, + format!("Unknown Transaction UID {transaction_uid}"), + ) + .into_response(), + Some(CommitmentState::Pending) => StatusCode::ACCEPTED.into_response(), + Some(CommitmentState::Completed(result)) => { + let json = DicomJson::from(InMemDicomObject::from(result)); + + Response::builder() + .status(StatusCode::OK) + .header(CONTENT_TYPE, mime::APPLICATION_JSON.as_ref()) + .body(Body::from(serde_json::to_string(&json).unwrap())) + .unwrap() + } + } +} diff --git a/src/api/stgcmt/service.rs b/src/api/stgcmt/service.rs new file mode 100644 index 0000000..f4f6ecc --- /dev/null +++ b/src/api/stgcmt/service.rs @@ -0,0 +1,183 @@ +use crate::api::stow::InstanceReference; +use crate::types::{UI, US}; +use async_trait::async_trait; +use axum::body::Body; +use axum::http::{Response, StatusCode}; +use axum::response::IntoResponse; +use dicom::core::value::{DataSetSequence, Value}; +use dicom::core::{DataElement, VR}; +use dicom::dicom_value; +use dicom::dictionary_std::tags; +use dicom::object::mem::InMemElement; +use dicom::object::InMemDicomObject; +use thiserror::Error; + +/// Storage Commitment Request Module. +/// +pub struct CommitRequest { + pub transaction_uid: UI, + pub referenced_sop_sequence: Vec, +} + +impl CommitRequest { + pub fn from_object( + transaction_uid: UI, + object: &InMemDicomObject, + ) -> Result { + let items = object + .get(tags::REFERENCED_SOP_SEQUENCE) + .and_then(InMemElement::items) + .ok_or(CommitError::MissingReferencedSopSequence)?; + + let referenced_sop_sequence = items + .iter() + .map(|item| { + let sop_class_uid = item + .get(tags::REFERENCED_SOP_CLASS_UID) + .map(InMemElement::to_str) + .and_then(Result::ok) + .ok_or(CommitError::MissingReferencedSopSequence)? + .into_owned(); + let sop_instance_uid = item + .get(tags::REFERENCED_SOP_INSTANCE_UID) + .map(InMemElement::to_str) + .and_then(Result::ok) + .ok_or(CommitError::MissingReferencedSopSequence)? + .into_owned(); + + Ok(InstanceReference { + sop_class_uid, + sop_instance_uid, + }) + }) + .collect::, CommitError>>()?; + + if referenced_sop_sequence.is_empty() { + return Err(CommitError::MissingReferencedSopSequence); + } + + Ok(Self { + transaction_uid, + referenced_sop_sequence, + }) + } +} + +/// An instance for which storage has not been committed. +#[derive(Debug, Clone)] +pub struct FailedInstance { + pub reference: InstanceReference, + pub failure_reason: US, +} + +/// The result of a storage commitment request, once known. +/// Storage Commitment Response Module. +/// +#[derive(Debug, Clone, Default)] +pub struct CommitmentResult { + pub referenced_sequence: Vec, + pub failed_sequence: Vec, +} + +impl From for InMemDicomObject { + fn from(result: CommitmentResult) -> Self { + let mut object = Self::new_empty(); + + let mut referenced_sequence = InMemElement::new( + tags::REFERENCED_SOP_SEQUENCE, + VR::SQ, + Value::Sequence(DataSetSequence::empty()), + ); + let referenced_items = referenced_sequence.items_mut().expect("Sequence exists"); + for referenced in result.referenced_sequence { + referenced_items.push(Self::from_element_iter([ + DataElement::new( + tags::REFERENCED_SOP_CLASS_UID, + VR::UI, + dicom_value!(Str, referenced.sop_class_uid), + ), + DataElement::new( + tags::REFERENCED_SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, referenced.sop_instance_uid), + ), + ])); + } + + let mut failed_sequence = InMemElement::new( + tags::FAILED_SOP_SEQUENCE, + VR::SQ, + Value::Sequence(DataSetSequence::empty()), + ); + let failed_items = failed_sequence.items_mut().expect("Sequence exists"); + for failed in result.failed_sequence { + failed_items.push(Self::from_element_iter([ + DataElement::new( + tags::REFERENCED_SOP_CLASS_UID, + VR::UI, + dicom_value!(Str, failed.reference.sop_class_uid), + ), + DataElement::new( + tags::REFERENCED_SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, failed.reference.sop_instance_uid), + ), + DataElement::new( + tags::FAILURE_REASON, + VR::US, + dicom_value!(U16, [failed.failure_reason]), + ), + ])); + } + + object.put(referenced_sequence); + object.put(failed_sequence); + object + } +} + +/// The state of a previously submitted commitment request. +#[derive(Debug, Clone)] +pub enum CommitmentState { + /// The origin server has not finished processing the storage commitment request yet. + Pending, + /// The origin server finished processing the storage commitment request. + Completed(CommitmentResult), +} + +/// +#[async_trait] +pub trait StgcmtService: Sync + Send { + /// Commit Transaction. + /// + async fn commit(&self, request: CommitRequest) -> Result<(), CommitError>; + + /// Check Commit Result Transaction. + /// + async fn check_result(&self, transaction_uid: &str) -> Option; +} + +#[derive(Debug, Error)] +pub enum CommitError { + #[error("A commitment request with Transaction UID {0} already exists")] + DuplicateTransaction(UI), + #[error("The request payload did not contain a (non-empty) Referenced SOP Sequence")] + MissingReferencedSopSequence, + #[error("Failed to send N-ACTION-RQ: {0:#}")] + Backend(#[from] anyhow::Error), +} + +impl IntoResponse for CommitError { + fn into_response(self) -> Response { + let status = match &self { + Self::DuplicateTransaction(_) => StatusCode::CONFLICT, + Self::MissingReferencedSopSequence => StatusCode::BAD_REQUEST, + Self::Backend(_) => StatusCode::SERVICE_UNAVAILABLE, + }; + + Response::builder() + .status(status) + .body(Body::from(self.to_string())) + .unwrap() + } +} diff --git a/src/api/stow/service.rs b/src/api/stow/service.rs index d1638f4..1de6d17 100644 --- a/src/api/stow/service.rs +++ b/src/api/stow/service.rs @@ -16,7 +16,7 @@ pub struct StoreRequest { } /// -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct InstanceReference { pub sop_class_uid: UI, pub sop_instance_uid: UI, diff --git a/src/backend/dimse/cstore/storescp.rs b/src/backend/dimse/cstore/storescp.rs index 2a18395..7d7bd84 100644 --- a/src/backend/dimse/cstore/storescp.rs +++ b/src/backend/dimse/cstore/storescp.rs @@ -5,7 +5,11 @@ use crate::backend::dimse::cmove::{ use crate::backend::dimse::cstore::{ CompositeStoreResponse, COMMAND_FIELD_COMPOSITE_STORE_REQUEST, }; -use crate::backend::dimse::{DicomMessageReader, DicomMessageWriter}; +use crate::backend::dimse::stgcmt::store::StorageCommitmentStore; +use crate::backend::dimse::stgcmt::{ + EventReportRequest, EventReportResponse, COMMAND_FIELD_N_EVENT_REPORT_REQUEST, +}; +use crate::backend::dimse::{DicomMessage, DicomMessageReader, DicomMessageWriter}; use crate::config::DimseServerConfig; use crate::types::{AE, UI, US}; use anyhow::Context; @@ -27,15 +31,22 @@ pub struct StoreServiceClassProvider { struct InnerStoreServiceClassProvider { mediator: MoveMediator, subscribers: Vec, + stgcmt_store: StorageCommitmentStore, config: DimseServerConfig, } impl StoreServiceClassProvider { - pub fn new(mediator: MoveMediator, subscribers: Vec, config: DimseServerConfig) -> Self { + pub fn new( + mediator: MoveMediator, + subscribers: Vec, + stgcmt_store: StorageCommitmentStore, + config: DimseServerConfig, + ) -> Self { Self { inner: Arc::new(InnerStoreServiceClassProvider { mediator, subscribers, + stgcmt_store, config, }), } @@ -108,9 +119,14 @@ impl StoreServiceClassProvider { .and_then(Result::ok) .context("Missing tag COMMAND_FIELD (0000,0100)")?; + if command_field == COMMAND_FIELD_N_EVENT_REPORT_REQUEST { + Self::handle_event_report(&association, message, &inner).await?; + continue; + } + if command_field != COMMAND_FIELD_COMPOSITE_STORE_REQUEST { return Err(anyhow::Error::msg( - "Unexpected Command Field. Only C-STORE-RQ is supported.", + "Unexpected Command Field. Only C-STORE-RQ and N-EVENT-REPORT-RQ are supported.", )); } @@ -187,4 +203,38 @@ impl StoreServiceClassProvider { } Ok(()) } + + /// Handles an inbound N-EVENT-REPORT-RQ, which reports the (asynchronous) result of a + /// previously sent Storage Commitment N-ACTION-RQ. + /// See [`crate::backend::dimse::stgcmt::scu::StorageCommitmentServiceClassUser`]. + async fn handle_event_report( + association: &ServerAssociation, + message: DicomMessage, + inner: &InnerStoreServiceClassProvider, + ) -> anyhow::Result<()> { + let presentation_context_id = message.presentation_context_id; + let request = EventReportRequest::try_from(message)?; + + info!( + transaction_uid = request.transaction_uid, + event_type_id = request.event_type_id, + successes = request.result.referenced_sequence.len(), + failures = request.result.failed_sequence.len(), + "Received N-EVENT-REPORT-RQ" + ); + + inner + .stgcmt_store + .complete(request.transaction_uid, request.result); + + let response = EventReportResponse { + message_id: request.message_id, + }; + + association + .write_message(response, presentation_context_id, Duration::from_secs(10)) + .await?; + + Ok(()) + } } diff --git a/src/backend/dimse/mod.rs b/src/backend/dimse/mod.rs index 7973e11..6f1e037 100644 --- a/src/backend/dimse/mod.rs +++ b/src/backend/dimse/mod.rs @@ -4,6 +4,9 @@ //! It depends on a store service class provider that must run in the background. //! - STOR-RS is implemented as a store service class user (C-STORE service). //! - MWL-RS is implemented as a find service class user (C-FIND service). +//! - Storage Commitment is implemented as an N-ACTION service class user. +//! The N-EVENT-REPORT that reports the result is received by the same store service class +//! provider that receives C-STORE requests. //! mod cecho; @@ -14,6 +17,7 @@ mod cstore; pub mod association; pub mod mwl; pub mod qido; +pub mod stgcmt; pub mod stow; pub mod wado; diff --git a/src/backend/dimse/stgcmt/mod.rs b/src/backend/dimse/stgcmt/mod.rs new file mode 100644 index 0000000..9340a77 --- /dev/null +++ b/src/backend/dimse/stgcmt/mod.rs @@ -0,0 +1,251 @@ +use crate::api::stgcmt::{CommitmentResult, FailedInstance}; +use crate::api::stow::InstanceReference; +use crate::backend::dimse::{DicomMessage, ReadError, DATA_SET_EXISTS, DATA_SET_MISSING}; +use crate::types::{UI, US}; +use dicom::core::value::{DataSetSequence, Value}; +use dicom::core::{DataElement, Tag, VR}; +use dicom::dicom_value; +use dicom::dictionary_std::{tags, uids}; +use dicom::object::mem::InMemElement; +use dicom::object::InMemDicomObject; + +pub mod scu; +pub mod service; +pub mod store; + +pub use service::DimseStgcmtService; + +// Magic numbers defined by the DICOM specification. +pub const COMMAND_FIELD_N_ACTION_REQUEST: US = 0x0130; +#[allow(unused)] +pub const COMMAND_FIELD_N_ACTION_RESPONSE: US = 0x8130; +pub const COMMAND_FIELD_N_EVENT_REPORT_REQUEST: US = 0x0100; +pub const COMMAND_FIELD_N_EVENT_REPORT_RESPONSE: US = 0x8100; + +pub const ACTION_TYPE_ID_STORAGE_COMMITMENT_REQUEST: US = 1; +/// +#[allow(unused)] +pub const EVENT_TYPE_ID_STORAGE_COMMITMENT_COMPLETE_FAILURES_EXIST: US = 2; + +fn referenced_sop_sequence_element(instances: &[InstanceReference]) -> InMemElement { + let mut element = InMemElement::new( + tags::REFERENCED_SOP_SEQUENCE, + VR::SQ, + Value::Sequence(DataSetSequence::empty()), + ); + let items = element.items_mut().expect("Sequence exists"); + for instance in instances { + items.push(InMemDicomObject::from_element_iter([ + DataElement::new( + tags::REFERENCED_SOP_CLASS_UID, + VR::UI, + dicom_value!(Str, instance.sop_class_uid.clone()), + ), + DataElement::new( + tags::REFERENCED_SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, instance.sop_instance_uid.clone()), + ), + ])); + } + element +} + +fn parse_referenced_sop_sequence(object: &InMemDicomObject, tag: Tag) -> Vec { + object + .get(tag) + .and_then(InMemElement::items) + .map(|items| { + items + .iter() + .filter_map(|item| { + let sop_class_uid = item + .get(tags::REFERENCED_SOP_CLASS_UID) + .map(InMemElement::to_str) + .and_then(Result::ok)? + .into_owned(); + let sop_instance_uid = item + .get(tags::REFERENCED_SOP_INSTANCE_UID) + .map(InMemElement::to_str) + .and_then(Result::ok)? + .into_owned(); + Some(InstanceReference { + sop_class_uid, + sop_instance_uid, + }) + }) + .collect() + }) + .unwrap_or_default() +} + +fn parse_failed_sop_sequence(object: &InMemDicomObject) -> Vec { + object + .get(tags::FAILED_SOP_SEQUENCE) + .and_then(InMemElement::items) + .map(|items| { + items + .iter() + .filter_map(|item| { + let sop_class_uid = item + .get(tags::REFERENCED_SOP_CLASS_UID) + .map(InMemElement::to_str) + .and_then(Result::ok)? + .into_owned(); + let sop_instance_uid = item + .get(tags::REFERENCED_SOP_INSTANCE_UID) + .map(InMemElement::to_str) + .and_then(Result::ok)? + .into_owned(); + let failure_reason = item + .get(tags::FAILURE_REASON) + .map(InMemElement::to_int::) + .and_then(Result::ok) + .unwrap_or_default(); + Some(FailedInstance { + reference: InstanceReference { + sop_class_uid, + sop_instance_uid, + }, + failure_reason, + }) + }) + .collect() + }) + .unwrap_or_default() +} + +/// N-ACTION-RQ +/// +pub struct NActionRequest { + pub message_id: US, + pub transaction_uid: UI, + pub referenced_sop_sequence: Vec, +} + +impl From for DicomMessage { + #[rustfmt::skip] + fn from(request: NActionRequest) -> Self { + let command = InMemDicomObject::command_from_element_iter([ + DataElement::new(tags::AFFECTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::STORAGE_COMMITMENT_PUSH_MODEL)), + DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_N_ACTION_REQUEST])), + DataElement::new(tags::MESSAGE_ID, VR::US, dicom_value!(U16, [request.message_id])), + DataElement::new(tags::REQUESTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::STORAGE_COMMITMENT_PUSH_MODEL)), + DataElement::new(tags::REQUESTED_SOP_INSTANCE_UID, VR::UI, dicom_value!(Str, uids::STORAGE_COMMITMENT_PUSH_MODEL_INSTANCE)), + DataElement::new(tags::ACTION_TYPE_ID, VR::US, dicom_value!(U16, [ACTION_TYPE_ID_STORAGE_COMMITMENT_REQUEST])), + DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_EXISTS])), + ]); + + let mut data_set = InMemDicomObject::from_element_iter([ + DataElement::new(tags::TRANSACTION_UID, VR::UI, dicom_value!(Str, request.transaction_uid)), + ]); + data_set.put(referenced_sop_sequence_element(&request.referenced_sop_sequence)); + + Self { + command, + data: Some(data_set), + presentation_context_id: None, + } + } +} + +/// N-ACTION-RSP +#[derive(Debug)] +pub struct NActionResponse { + pub status: US, +} + +impl TryFrom for NActionResponse { + type Error = ReadError; + + fn try_from(message: DicomMessage) -> Result { + let status = message + .command + .get(tags::STATUS) + .map(InMemElement::to_int::) + .and_then(Result::ok) + .ok_or(ReadError::MissingAttribute(tags::STATUS))?; + + Ok(Self { status }) + } +} + +/// N-EVENT-REPORT-RQ, as received by the [`crate::backend::dimse::StoreServiceClassProvider`]. +#[derive(Debug)] +pub struct EventReportRequest { + pub message_id: US, + pub event_type_id: US, + pub transaction_uid: UI, + pub result: CommitmentResult, +} + +impl TryFrom for EventReportRequest { + type Error = ReadError; + + fn try_from(message: DicomMessage) -> Result { + let message_id = message + .command + .get(tags::MESSAGE_ID) + .map(InMemElement::to_int::) + .and_then(Result::ok) + .ok_or(ReadError::MissingAttribute(tags::MESSAGE_ID))?; + + let event_type_id = message + .command + .get(tags::EVENT_TYPE_ID) + .map(InMemElement::to_int::) + .and_then(Result::ok) + .ok_or(ReadError::MissingAttribute(tags::EVENT_TYPE_ID))?; + + let data = message + .data + .ok_or(ReadError::MissingAttribute(tags::TRANSACTION_UID))?; + + let transaction_uid = data + .get(tags::TRANSACTION_UID) + .map(InMemElement::to_str) + .and_then(Result::ok) + .ok_or(ReadError::MissingAttribute(tags::TRANSACTION_UID))? + .into_owned(); + + let result = CommitmentResult { + referenced_sequence: parse_referenced_sop_sequence( + &data, + tags::REFERENCED_SOP_SEQUENCE, + ), + failed_sequence: parse_failed_sop_sequence(&data), + }; + + Ok(Self { + message_id, + event_type_id, + transaction_uid, + result, + }) + } +} + +/// N-EVENT-REPORT-RSP +pub struct EventReportResponse { + pub message_id: US, +} + +impl From for DicomMessage { + #[rustfmt::skip] + fn from(response: EventReportResponse) -> Self { + let command = InMemDicomObject::command_from_element_iter([ + DataElement::new(tags::AFFECTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::STORAGE_COMMITMENT_PUSH_MODEL)), + DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_N_EVENT_REPORT_RESPONSE])), + DataElement::new(tags::MESSAGE_ID_BEING_RESPONDED_TO, VR::US, dicom_value!(U16, [response.message_id])), + DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_MISSING])), + DataElement::new(tags::STATUS, VR::US, dicom_value!(U16, [0u16])), + DataElement::new(tags::AFFECTED_SOP_INSTANCE_UID, VR::UI, dicom_value!(Str, uids::STORAGE_COMMITMENT_PUSH_MODEL_INSTANCE)), + ]); + + Self { + command, + data: None, + presentation_context_id: None, + } + } +} diff --git a/src/backend/dimse/stgcmt/scu.rs b/src/backend/dimse/stgcmt/scu.rs new file mode 100644 index 0000000..13d1874 --- /dev/null +++ b/src/backend/dimse/stgcmt/scu.rs @@ -0,0 +1,75 @@ +use crate::api::stow::InstanceReference; +use crate::backend::dimse::association; +use crate::backend::dimse::stgcmt::{NActionRequest, NActionResponse}; +use crate::backend::dimse::{ + DicomMessageReader, DicomMessageWriter, ReadError, StatusType, WriteError, +}; +use crate::types::{UI, US}; +use association::pool::{AssociationPool, PoolError, PresentationParameter}; +use association::AssociationError; +use dicom::dictionary_std::uids; +use std::time::Duration; +use thiserror::Error; +use tracing::trace; + +pub struct StorageCommitmentServiceClassUser { + pool: AssociationPool, + timeout: Duration, +} + +impl StorageCommitmentServiceClassUser { + pub const fn new(pool: AssociationPool, timeout: Duration) -> Self { + Self { pool, timeout } + } + + /// Sends an N-ACTION-RQ requesting storage commitment for the given instances. + /// Returns once the N-ACTION-RSP (a simple acknowledgement) has been received - + /// the actual commitment result arrives later, out-of-band, as an N-EVENT-REPORT-RQ. + #[allow(clippy::significant_drop_tightening)] + pub async fn commit( + &self, + message_id: US, + transaction_uid: UI, + referenced_sop_sequence: Vec, + ) -> Result<(), CommitError> { + let association = self + .pool + .get(PresentationParameter { + abstract_syntax_uid: UI::from(uids::STORAGE_COMMITMENT_PUSH_MODEL), + transfer_syntax_uids: vec![UI::from(uids::IMPLICIT_VR_LITTLE_ENDIAN)], + }) + .await?; + + let request = NActionRequest { + message_id, + transaction_uid, + referenced_sop_sequence, + }; + + association + .write_message(request, None, self.timeout) + .await?; + trace!("Sent N-ACTION-RQ"); + + let message = association.read_message(self.timeout).await?; + trace!("Received N-ACTION-RSP"); + + let response = NActionResponse::try_from(message)?; + match StatusType::try_from(response.status) { + Ok(StatusType::Success) => Ok(()), + _ => Err(CommitError::Failure(response.status)), + } + } +} + +#[derive(Debug, Error)] +pub enum CommitError { + #[error(transparent)] + Read(#[from] ReadError), + #[error(transparent)] + Write(#[from] WriteError), + #[error(transparent)] + Association(#[from] PoolError), + #[error("N-ACTION-RSP indicated failure (status 0x{0:04X})")] + Failure(US), +} diff --git a/src/backend/dimse/stgcmt/service.rs b/src/backend/dimse/stgcmt/service.rs new file mode 100644 index 0000000..9a1eaea --- /dev/null +++ b/src/backend/dimse/stgcmt/service.rs @@ -0,0 +1,54 @@ +use crate::api::stgcmt::{CommitError, CommitRequest, CommitmentState, StgcmtService}; +use crate::backend::dimse::association::pool::AssociationPool; +use crate::backend::dimse::next_message_id; +use crate::backend::dimse::stgcmt::scu::StorageCommitmentServiceClassUser; +use crate::backend::dimse::stgcmt::store::StorageCommitmentStore; +use async_trait::async_trait; +use std::time::Duration; + +pub struct DimseStgcmtService { + scu: StorageCommitmentServiceClassUser, + store: StorageCommitmentStore, +} + +impl DimseStgcmtService { + pub const fn new( + pool: AssociationPool, + timeout: Duration, + store: StorageCommitmentStore, + ) -> Self { + let scu = StorageCommitmentServiceClassUser::new(pool, timeout); + Self { scu, store } + } +} + +#[async_trait] +impl StgcmtService for DimseStgcmtService { + async fn commit(&self, request: CommitRequest) -> Result<(), CommitError> { + self.store + .insert_pending(request.transaction_uid.clone()) + .map_err(|_| CommitError::DuplicateTransaction(request.transaction_uid.clone()))?; + + let result = self + .scu + .commit( + next_message_id(), + request.transaction_uid.clone(), + request.referenced_sop_sequence, + ) + .await; + + if let Err(err) = result { + // The N-ACTION-RQ itself failed synchronously, so no N-EVENT-REPORT-RQ will ever + // arrive for this Transaction UID - allow the client to retry. + self.store.remove(&request.transaction_uid); + return Err(CommitError::Backend(err.into())); + } + + Ok(()) + } + + async fn check_result(&self, transaction_uid: &str) -> Option { + self.store.get(transaction_uid) + } +} diff --git a/src/backend/dimse/stgcmt/store.rs b/src/backend/dimse/stgcmt/store.rs new file mode 100644 index 0000000..3f3ab43 --- /dev/null +++ b/src/backend/dimse/stgcmt/store.rs @@ -0,0 +1,73 @@ +use crate::api::stgcmt::{CommitmentResult, CommitmentState}; +use crate::types::UI; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +/// Correlates a Storage Commitment N-ACTION-RQ (sent by [`super::scu::StorageCommitmentServiceClassUser`]) +/// with the N-EVENT-REPORT-RQ that later, asynchronously, reports its result +/// (received by [`crate::backend::dimse::StoreServiceClassProvider`]), keyed by Transaction UID. +/// +/// This is intentionally a plain map rather than the callback-based +/// [`crate::backend::dimse::cmove::MoveMediator`] pattern: the Commit Transaction (POST) and the +/// Check Commit Result Transaction (GET) that later reads this store are two independent HTTP +/// requests, arbitrarily far apart in time, with no single task blocked waiting for a callback. +/// +/// Like [`crate::backend::dimse::association::pool::AssociationPools`] and +/// [`crate::backend::dimse::cmove::MoveMediator`], this store is memory-only: a commitment result +/// delivered while the process is restarted between the N-ACTION and the N-EVENT-REPORT is lost +/// unless the peer retries delivery. +#[derive(Clone, Default)] +pub struct StorageCommitmentStore { + inner: Arc>>, +} + +/// Returned by [`StorageCommitmentStore::insert_pending`] when the given Transaction UID has +/// already been submitted. +#[derive(Debug)] +pub struct DuplicateTransaction; + +impl StorageCommitmentStore { + pub fn new() -> Self { + Self::default() + } + + /// Registers a new, still-unresolved commitment request. + /// Fails if the Transaction UID is already known, per the Commit Transaction's + /// 409 (Conflict) status code. + pub fn insert_pending(&self, transaction_uid: UI) -> Result<(), DuplicateTransaction> { + let is_duplicate = { + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + let is_duplicate = states.contains_key(&transaction_uid); + if !is_duplicate { + states.insert(transaction_uid, CommitmentState::Pending); + } + is_duplicate + }; + + if is_duplicate { + return Err(DuplicateTransaction); + } + Ok(()) + } + + /// Records the result of a commitment request, once its N-EVENT-REPORT-RQ has arrived. + /// Upserts unconditionally, even without a matching `Pending` entry (e.g. after a restart), + /// so a client that keeps polling can still observe the result once it arrives. + pub fn complete(&self, transaction_uid: UI, result: CommitmentResult) { + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + states.insert(transaction_uid, CommitmentState::Completed(result)); + } + + pub fn get(&self, transaction_uid: &str) -> Option { + let states = self.inner.lock().expect("mutex should not be poisoned"); + states.get(transaction_uid).cloned() + } + + /// Removes a pending entry, e.g. after the N-ACTION-RQ itself failed synchronously + /// (in which case no N-EVENT-REPORT-RQ will ever arrive), so the Transaction UID can be + /// retried. + pub fn remove(&self, transaction_uid: &str) { + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + states.remove(transaction_uid); + } +} diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 973e37c..726a0af 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -1,5 +1,6 @@ use crate::api::mwl::MwlService; use crate::api::qido::QidoService; +use crate::api::stgcmt::StgcmtService; use crate::api::stow::StowService; use crate::api::wado::WadoService; use crate::config::BackendConfig; @@ -20,6 +21,7 @@ pub struct ServiceProvider { pub wado: Option>, pub stow: Option>, pub mwl: Option>, + pub stgcmt: Option>, } impl FromRequestParts for ServiceProvider @@ -53,6 +55,7 @@ where BackendConfig::Dimse { .. } => { use crate::backend::dimse::mwl::DimseMwlService; use crate::backend::dimse::qido::DimseQidoService; + use crate::backend::dimse::stgcmt::DimseStgcmtService; use crate::backend::dimse::stow::DimseStowService; use crate::backend::dimse::wado::DimseWadoService; @@ -77,6 +80,11 @@ where pool.to_owned(), Duration::from_millis(ae_config.mwl.timeout), ))), + stgcmt: Some(Box::new(DimseStgcmtService::new( + pool.to_owned(), + Duration::from_millis(ae_config.stgcmt.timeout), + state.stgcmt_store, + ))), } } #[cfg(feature = "s3")] @@ -88,6 +96,7 @@ where wado: Some(Box::new(S3WadoService::new(&config))), stow: None, mwl: None, + stgcmt: None, } } }; diff --git a/src/config/mod.rs b/src/config/mod.rs index d7047ef..0ddd47c 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -31,6 +31,8 @@ pub struct ApplicationEntityConfig { pub stow: StowConfig, #[serde(default, rename = "mwl-rs")] pub mwl: MwlConfig, + #[serde(default, rename = "stgcmt-rs")] + pub stgcmt: StgcmtConfig, } #[derive(Debug, Clone, Deserialize)] @@ -195,6 +197,20 @@ impl Default for MwlConfig { } } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct StgcmtConfig { + /// Timeout for the N-ACTION-RQ/RSP round trip. The actual commitment result is delivered + /// later, out-of-band, as an N-EVENT-REPORT-RQ, and is not subject to this timeout. + pub timeout: u64, +} + +impl Default for StgcmtConfig { + fn default() -> Self { + Self { timeout: 30_000 } + } +} + impl AppConfig { /// Loads the application configuration from the following sources: /// 1. Defaults (defined in `defaults.toml`) diff --git a/src/main.rs b/src/main.rs index 6c23b68..46b975c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ pub(crate) mod utils; use crate::backend::dimse::association; use crate::backend::dimse::cmove::MoveMediator; +use crate::backend::dimse::stgcmt::store::StorageCommitmentStore; use crate::backend::dimse::StoreServiceClassProvider; use crate::config::{AppConfig, HttpServerConfig}; use crate::types::AE; @@ -64,6 +65,7 @@ pub struct AppState { pub config: AppConfig, pub pools: AssociationPools, pub mediator: MoveMediator, + pub stgcmt_store: StorageCommitmentStore, } fn init_sentry(config: &AppConfig) -> sentry::ClientInitGuard { @@ -107,15 +109,18 @@ fn main() -> Result<(), Box> { async fn run(config: AppConfig) -> anyhow::Result<()> { let mediator = MoveMediator::new(&config); let pools = AssociationPools::new(&config); + let stgcmt_store = StorageCommitmentStore::new(); let app_state = AppState { config: config.clone(), mediator: mediator.clone(), pools, + stgcmt_store: stgcmt_store.clone(), }; for dimse_config in config.server.dimse { let mediator = mediator.clone(); + let stgcmt_store = stgcmt_store.clone(); let subscribers: Vec = config .aets .iter() @@ -125,7 +130,8 @@ async fn run(config: AppConfig) -> anyhow::Result<()> { .collect(); tokio::spawn(async move { - let storescp = StoreServiceClassProvider::new(mediator, subscribers, dimse_config); + let storescp = + StoreServiceClassProvider::new(mediator, subscribers, stgcmt_store, dimse_config); if let Err(err) = storescp.spawn().await { error!("Failed to spawn STORE-SCP thread: {err}"); // Unrecoverable error - exit the process diff --git a/src/utils/dicom_json.rs b/src/utils/dicom_json.rs new file mode 100644 index 0000000..6315949 --- /dev/null +++ b/src/utils/dicom_json.rs @@ -0,0 +1,45 @@ +use axum::body::Bytes; +use axum::extract::{FromRequest, Request}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use dicom::object::InMemDicomObject; + +/// Extracts an [`InMemDicomObject`] from a request body encoded as DICOM JSON. +/// +pub struct DicomJsonBody(pub InMemDicomObject); + +pub enum DicomJsonBodyRejection { + InvalidBody(axum::extract::rejection::BytesRejection), + InvalidJson(serde_json::Error), +} + +impl IntoResponse for DicomJsonBodyRejection { + fn into_response(self) -> Response { + match self { + Self::InvalidBody(err) => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), + Self::InvalidJson(err) => ( + StatusCode::BAD_REQUEST, + format!("Failed to parse DICOM JSON payload: {err}"), + ) + .into_response(), + } + } +} + +impl FromRequest for DicomJsonBody +where + S: Send + Sync, +{ + type Rejection = DicomJsonBodyRejection; + + async fn from_request(request: Request, state: &S) -> Result { + let bytes = Bytes::from_request(request, state) + .await + .map_err(DicomJsonBodyRejection::InvalidBody)?; + + let object: InMemDicomObject = + dicom_json::from_slice(&bytes).map_err(DicomJsonBodyRejection::InvalidJson)?; + + Ok(Self(object)) + } +} diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 30ba0c2..fc2bc47 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1 +1,2 @@ +pub mod dicom_json; pub mod multipart; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f4fb392..c6ec300 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1,11 +1,11 @@ -use anyhow::{bail, Context}; +use anyhow::Context; use dicom_web::DicomWebClient; use std::path::PathBuf; use std::process::Stdio; use std::time::Duration; -use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::core::{Host, IntoContainerPort, WaitFor}; use testcontainers::runners::AsyncRunner; -use testcontainers::{ContainerAsync, GenericImage}; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; use tokio::io::{AsyncBufReadExt, BufReader, Lines}; use tokio::process::{Child, ChildStdout, Command}; @@ -14,6 +14,9 @@ pub async fn spawn_orthanc() -> anyhow::Result> { .with_exposed_port(4242.tcp()) .with_exposed_port(8042.tcp()) .with_wait_for(WaitFor::message_on_stderr("Orthanc has started")) + // Allows the Orthanc container to dial back into a DICOM-RST process running on the + // test host, e.g. to deliver a Storage Commitment N-EVENT-REPORT-RQ. + .with_host("host.docker.internal", Host::HostGateway) .start() .await .context("failed to start Orthanc container") @@ -21,7 +24,9 @@ pub async fn spawn_orthanc() -> anyhow::Result> { pub async fn spawn_dicomrst(config: &str) -> anyhow::Result { let mut server = ServerProcess::spawn(config)?; - server.http_port = server.wait_until_started().await?; + let (http_port, dimse_port) = server.wait_until_started().await?; + server.http_port = http_port; + server.dimse_port = dimse_port; Ok(server) } @@ -29,7 +34,8 @@ pub struct ServerProcess { child: Child, stdout: Lines>, workdir: PathBuf, - http_port: u16, + pub http_port: u16, + pub dimse_port: u16, } impl ServerProcess { @@ -53,36 +59,41 @@ impl ServerProcess { stdout, workdir, http_port: 0, + dimse_port: 0, }) } - async fn wait_until_started(&mut self) -> anyhow::Result { + fn parse_port(line: &str) -> anyhow::Result { + line.split_whitespace() + .find_map(|part| part.strip_prefix("server.port=")) + .ok_or_else(|| anyhow::Error::msg("Log line did not contain server.port="))? + .parse::() + .context("Failed to parse server.port as u16") + } + + /// Waits until DICOM-RST has logged both its HTTP and DIMSE listener ports, returning + /// `(http_port, dimse_port)`. + async fn wait_until_started(&mut self) -> anyhow::Result<(u16, u16)> { tokio::time::timeout(Duration::from_secs(15), async { - while let Some(line) = self - .stdout - .next_line() - .await - .context("Failed to read DICOM-RST stdout")? - { - if !line.contains("Started DICOMweb server") { - continue; + let mut http_port = None; + let mut dimse_port = None; + + while http_port.is_none() || dimse_port.is_none() { + let line = self + .stdout + .next_line() + .await + .context("Failed to read DICOM-RST stdout")? + .context("DICOM-RST exited before becoming ready")?; + + if line.contains("Started DICOMweb server") { + http_port = Some(Self::parse_port(&line)?); + } else if line.contains("Started Store Service Class Provider") { + dimse_port = Some(Self::parse_port(&line)?); } - - let port = line - .split_whitespace() - .find_map(|part| part.strip_prefix("server.port=")) - .ok_or_else(|| { - anyhow::Error::msg( - "DICOM-RST started, but stdout did not contain server.port=", - ) - })? - .parse::() - .context("Failed to parse DICOM-RST server.port as u16")?; - - return Ok(port); } - bail!("DICOM-RST exited before becoming ready"); + Ok((http_port.unwrap(), dimse_port.unwrap())) }) .await .context("Timed out waiting for DICOM-RST to start")? diff --git a/tests/stgcmt.rs b/tests/stgcmt.rs new file mode 100644 index 0000000..a7d7947 --- /dev/null +++ b/tests/stgcmt.rs @@ -0,0 +1,223 @@ +mod common; + +use anyhow::Context; +use common::*; +use dicom::core::value::{DataSetSequence, Value}; +use dicom::core::{DataElement, VR}; +use dicom::dicom_value; +use dicom::dictionary_std::tags; +use dicom::object::mem::InMemElement; +use dicom::object::{open_file, InMemDicomObject}; +use dicom_web::DicomWebClient; +use futures::stream; +use std::time::Duration; +use testcontainers::core::IntoContainerPort; + +/// Builds a DICOM JSON Storage Commitment Request Module (PS3.18 Annex J.1) referencing a single +/// SOP Instance. +fn commit_request_body(sop_class_uid: &str, sop_instance_uid: &str) -> anyhow::Result { + let mut sequence = InMemElement::new( + tags::REFERENCED_SOP_SEQUENCE, + VR::SQ, + Value::Sequence(DataSetSequence::empty()), + ); + sequence + .items_mut() + .expect("Sequence exists") + .push(InMemDicomObject::from_element_iter([ + DataElement::new( + tags::REFERENCED_SOP_CLASS_UID, + VR::UI, + dicom_value!(Str, sop_class_uid), + ), + DataElement::new( + tags::REFERENCED_SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, sop_instance_uid), + ), + ])); + + let mut object = InMemDicomObject::new_empty(); + object.put(sequence); + Ok(dicom_json::to_string(&object)?) +} + +fn new_transaction_uid() -> String { + format!("2.25.{}", uuid::Uuid::new_v4().as_u128()) +} + +/// Orthanc only accepts N-ACTION requests from AEs that it knows about, unlike its permissive +/// default for C-STORE - register DICOM-RST as a modality that Orthanc can dial back into (via +/// the host-gateway) to deliver the N-EVENT-REPORT-RQ. +async fn register_dicom_rst_as_modality( + http: &reqwest::Client, + orthanc_http_port: u16, + dimse_port: u16, +) -> anyhow::Result<()> { + http.put(format!( + "http://localhost:{orthanc_http_port}/modalities/DICOM-RST" + )) + .basic_auth("orthanc", Some("orthanc")) + .json(&serde_json::json!({ + "AET": "DICOM-RST", + "Host": "host.docker.internal", + "Port": dimse_port, + "AllowStorageCommitment": true, + })) + .send() + .await? + .error_for_status() + .context("failed to register DICOM-RST as an Orthanc modality")?; + Ok(()) +} + +/// Storage Commitment can only be requested for instances that already exist on the origin +/// server - STOWs a test instance and returns its (SOP Class UID, SOP Instance UID). +async fn stow_test_instance(http_port: u16) -> anyhow::Result<(String, String)> { + let instance = open_file(dicom_test_files::path("pydicom/CT_small.dcm").unwrap())?; + let sop_class_uid = instance.meta().media_storage_sop_class_uid().to_owned(); + let sop_instance_uid = instance.meta().media_storage_sop_instance_uid().to_owned(); + + let dicom_web = + DicomWebClient::with_single_url(&format!("http://localhost:{http_port}/aets/ORTHANC")); + dicom_web + .store_instances() + .with_instances(stream::iter([instance])) + .run() + .await + .context("STOW-RS request failed")?; + + Ok((sop_class_uid, sop_instance_uid)) +} + +#[tokio::test] +async fn can_commit_storage_and_check_result() -> anyhow::Result<()> { + let orthanc = spawn_orthanc().await?; + let orthanc_dimse_port = orthanc + .get_host_port_ipv4(4242.tcp()) + .await + .context("failed to get mapped Orthanc DIMSE port")?; + let orthanc_http_port = orthanc + .get_host_port_ipv4(8042.tcp()) + .await + .context("failed to get mapped Orthanc HTTP port")?; + + let config = format!( + " + server: + http: + port: 0 + dimse: + - aet: DICOM-RST + interface: 0.0.0.0 + port: 0 + aets: + - aet: ORTHANC + host: 127.0.0.1 + port: {orthanc_dimse_port} + backend: DIMSE + " + ); + let server = spawn_dicomrst(&config).await?; + let http = reqwest::Client::new(); + + register_dicom_rst_as_modality(&http, orthanc_http_port, server.dimse_port).await?; + let (sop_class_uid, sop_instance_uid) = stow_test_instance(server.http_port).await?; + + let base_url = format!( + "http://localhost:{}/aets/ORTHANC/commitment-requests", + server.http_port + ); + + // A commitment request for an instance that does not exist should end up in the + // FailedSOPSequence of the result, with a Failure Reason. + let failing_transaction_uid = new_transaction_uid(); + let response = http + .post(format!("{base_url}/{failing_transaction_uid}")) + .header("Content-Type", "application/dicom+json") + .body(commit_request_body( + &sop_class_uid, + "1.2.3.4.5.6.7.8.9.this-instance-does-not-exist", + )?) + .send() + .await?; + assert_eq!(response.status(), 202); + + // A commitment request for the instance that was just stored should succeed. + let transaction_uid = new_transaction_uid(); + let response = http + .post(format!("{base_url}/{transaction_uid}")) + .header("Content-Type", "application/dicom+json") + .body(commit_request_body(&sop_class_uid, &sop_instance_uid)?) + .send() + .await?; + assert_eq!(response.status(), 202); + + // Resubmitting the same Transaction UID while it is still pending must be rejected. + let response = http + .post(format!("{base_url}/{transaction_uid}")) + .header("Content-Type", "application/dicom+json") + .body(commit_request_body(&sop_class_uid, &sop_instance_uid)?) + .send() + .await?; + assert_eq!(response.status(), 409); + + // An unknown Transaction UID must be reported as such. + let response = http + .get(format!("{base_url}/does-not-exist")) + .send() + .await?; + assert_eq!(response.status(), 404); + + let result = poll_until_completed(&http, &format!("{base_url}/{transaction_uid}")).await?; + let referenced_sop_sequence = result + .get(tags::REFERENCED_SOP_SEQUENCE) + .context("Result is missing ReferencedSOPSequence")?; + assert!( + referenced_sop_sequence + .items() + .is_some_and(|items| items.len() == 1), + "Expected exactly one instance in ReferencedSOPSequence" + ); + let failed_sop_sequence = result + .get(tags::FAILED_SOP_SEQUENCE) + .context("Result is missing FailedSOPSequence")?; + assert!( + failed_sop_sequence.items().is_some_and(<[_]>::is_empty), + "Expected FailedSOPSequence to be empty" + ); + + let failing_result = + poll_until_completed(&http, &format!("{base_url}/{failing_transaction_uid}")).await?; + let failed_sop_sequence = failing_result + .get(tags::FAILED_SOP_SEQUENCE) + .context("Result is missing FailedSOPSequence")?; + assert!( + failed_sop_sequence + .items() + .is_some_and(|items| items.len() == 1), + "Expected exactly one instance in FailedSOPSequence" + ); + + Ok(()) +} + +/// Polls the Check Commit Result Transaction until it returns `200 OK`. +async fn poll_until_completed( + http: &reqwest::Client, + url: &str, +) -> anyhow::Result { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let response = http.get(url).send().await?; + if response.status() == 200 { + let body = response.text().await?; + return dicom_json::from_str::(&body) + .context("Failed to parse Storage Commitment Response Module"); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + }) + .await + .context("Timed out waiting for the Storage Commitment result")? +}