diff --git a/Cargo.lock b/Cargo.lock index ade1275..97e066f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1723,6 +1723,7 @@ dependencies = [ "mimalloc", "mime", "multer", + "reqwest 0.13.3", "sentry", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 9215765..75603a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,6 +65,7 @@ aws-credential-types = { version = "1.2.13", optional = true } testcontainers = "0.27.3" dicom-test-files = "0.4.0" dicom-web = "0.6.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..8bd56d8 --- /dev/null +++ b/src/api/stgcmt/service.rs @@ -0,0 +1,333 @@ +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, Tag, VR}; +use dicom::dicom_value; +use dicom::dictionary_std::tags; +use dicom::object::mem::InMemElement; +use dicom::object::InMemDicomObject; +use std::borrow::Cow; +use std::collections::HashMap; +use thiserror::Error; + +/// Storage Commitment Request Module. +/// +pub struct CommitRequest { + pub transaction_uid: UI, + /// All referenced SOP Instances, regardless of whether they were referenced by the + /// Referenced SOP Sequence or the Referenced Study Sequence. + pub instances: Vec, + /// See [`CommitmentResult::locations`]. + pub locations: HashMap, +} + +impl CommitRequest { + pub fn from_object( + transaction_uid: UI, + object: &InMemDicomObject, + ) -> Result { + let referenced_sop_sequence = object + .get(tags::REFERENCED_SOP_SEQUENCE) + .and_then(InMemElement::items); + let referenced_study_sequence = object + .get(tags::REFERENCED_STUDY_SEQUENCE) + .and_then(InMemElement::items); + + let mut instances = Vec::new(); + let mut locations = HashMap::new(); + + for item in referenced_sop_sequence.unwrap_or_default() { + instances.push(InstanceReference { + sop_class_uid: uid(item, tags::REFERENCED_SOP_CLASS_UID)?, + sop_instance_uid: uid(item, tags::REFERENCED_SOP_INSTANCE_UID)?, + }); + } + + for study in referenced_study_sequence.unwrap_or_default() { + let study_instance_uid = uid(study, tags::STUDY_INSTANCE_UID)?; + for series in items(study, tags::REFERENCED_SERIES_SEQUENCE)? { + let series_instance_uid = uid(series, tags::SERIES_INSTANCE_UID)?; + for sop_class in items(series, tags::REFERENCED_INSTANCES_BY_SOP_CLASS_SEQUENCE)? { + let sop_class_uid = uid(sop_class, tags::REFERENCED_SOP_CLASS_UID)?; + for instance in items(sop_class, tags::REFERENCED_INSTANCE_SEQUENCE)? { + let sop_instance_uid = uid(instance, tags::REFERENCED_SOP_INSTANCE_UID)?; + locations.insert( + sop_instance_uid.clone(), + InstanceLocation { + study_instance_uid: study_instance_uid.clone(), + series_instance_uid: series_instance_uid.clone(), + }, + ); + instances.push(InstanceReference { + sop_class_uid: sop_class_uid.clone(), + sop_instance_uid, + }); + } + } + } + } + + if instances.is_empty() { + return Err(CommitError::MissingReferences); + } + + Ok(Self { + transaction_uid, + instances, + locations, + }) + } +} + +/// Reads a required UID attribute of a request item. +fn uid(item: &InMemDicomObject, tag: Tag) -> Result { + item.get(tag) + .map(InMemElement::to_str) + .and_then(Result::ok) + .map(Cow::into_owned) + .ok_or(CommitError::MissingAttribute(tag)) +} + +/// Reads a required sequence attribute of a request item. +fn items(item: &InMemDicomObject, tag: Tag) -> Result<&[InMemDicomObject], CommitError> { + item.get(tag) + .and_then(InMemElement::items) + .ok_or(CommitError::MissingAttribute(tag)) +} + +/// The Study and Series of a SOP Instance that was referenced by the Referenced Study Sequence. +#[derive(Debug, Clone)] +pub struct InstanceLocation { + pub study_instance_uid: UI, + pub series_instance_uid: UI, +} + +/// 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, + /// The locations of the SOP Instances that were referenced by the Referenced Study Sequence + /// of the request, keyed by SOP Instance UID. The response references these instances by the + /// Referenced Study Sequence or Failed Study Sequence, all others by the Referenced SOP + /// Sequence or Failed SOP Sequence. + pub locations: HashMap, +} + +impl From for InMemDicomObject { + fn from(result: CommitmentResult) -> Self { + let mut referenced_items = Vec::new(); + let mut referenced_studies = StudyTree::default(); + for referenced in result.referenced_sequence { + if let Some(location) = result.locations.get(&referenced.sop_instance_uid) { + let instance = Self::from_element_iter([uid_element( + tags::REFERENCED_SOP_INSTANCE_UID, + referenced.sop_instance_uid, + )]); + referenced_studies.insert(location, &referenced.sop_class_uid, instance); + } else { + referenced_items.push(Self::from_element_iter([ + uid_element(tags::REFERENCED_SOP_CLASS_UID, referenced.sop_class_uid), + uid_element( + tags::REFERENCED_SOP_INSTANCE_UID, + referenced.sop_instance_uid, + ), + ])); + } + } + + let mut failed_items = Vec::new(); + let mut failed_studies = StudyTree::default(); + for failed in result.failed_sequence { + let failure_reason = DataElement::new( + tags::FAILURE_REASON, + VR::US, + dicom_value!(U16, [failed.failure_reason]), + ); + if let Some(location) = result.locations.get(&failed.reference.sop_instance_uid) { + let instance = Self::from_element_iter([ + uid_element( + tags::REFERENCED_SOP_INSTANCE_UID, + failed.reference.sop_instance_uid, + ), + failure_reason, + ]); + failed_studies.insert(location, &failed.reference.sop_class_uid, instance); + } else { + failed_items.push(Self::from_element_iter([ + uid_element( + tags::REFERENCED_SOP_CLASS_UID, + failed.reference.sop_class_uid, + ), + uid_element( + tags::REFERENCED_SOP_INSTANCE_UID, + failed.reference.sop_instance_uid, + ), + failure_reason, + ])); + } + } + + // All sequences are Type 1C: they are only present if at least one SOP Instance is + // referenced by them. + let mut object = Self::new_empty(); + if !referenced_items.is_empty() { + object.put(sequence_element( + tags::REFERENCED_SOP_SEQUENCE, + referenced_items, + )); + } + if !failed_items.is_empty() { + object.put(sequence_element(tags::FAILED_SOP_SEQUENCE, failed_items)); + } + if !referenced_studies.is_empty() { + object.put(referenced_studies.into_element(tags::REFERENCED_STUDY_SEQUENCE)); + } + if !failed_studies.is_empty() { + object.put(failed_studies.into_element(tags::FAILED_STUDY_SEQUENCE)); + } + object + } +} + +fn uid_element(tag: Tag, uid: UI) -> InMemElement { + DataElement::new(tag, VR::UI, dicom_value!(Str, uid)) +} + +fn sequence_element(tag: Tag, items: Vec) -> InMemElement { + InMemElement::new(tag, VR::SQ, Value::Sequence(DataSetSequence::from(items))) +} + +type Groups = Vec<(UI, T)>; + +/// Groups the instance items of a Referenced Study Sequence or Failed Study Sequence by +/// Study Instance UID, Series Instance UID and SOP Class UID. +#[derive(Default)] +struct StudyTree(Groups>>>); + +impl StudyTree { + fn insert( + &mut self, + location: &InstanceLocation, + sop_class_uid: &str, + instance: InMemDicomObject, + ) { + let series = group_mut(&mut self.0, &location.study_instance_uid); + let sop_classes = group_mut(series, &location.series_instance_uid); + group_mut(sop_classes, sop_class_uid).push(instance); + } + + const fn is_empty(&self) -> bool { + self.0.is_empty() + } + + fn into_element(self, tag: Tag) -> InMemElement { + let studies = self + .0 + .into_iter() + .map(|(study_instance_uid, series)| { + let series = series + .into_iter() + .map(|(series_instance_uid, sop_classes)| { + let sop_classes = sop_classes + .into_iter() + .map(|(sop_class_uid, instances)| { + InMemDicomObject::from_element_iter([ + uid_element(tags::REFERENCED_SOP_CLASS_UID, sop_class_uid), + sequence_element(tags::REFERENCED_INSTANCE_SEQUENCE, instances), + ]) + }) + .collect(); + InMemDicomObject::from_element_iter([ + uid_element(tags::SERIES_INSTANCE_UID, series_instance_uid), + sequence_element( + tags::REFERENCED_INSTANCES_BY_SOP_CLASS_SEQUENCE, + sop_classes, + ), + ]) + }) + .collect(); + InMemDicomObject::from_element_iter([ + uid_element(tags::STUDY_INSTANCE_UID, study_instance_uid), + sequence_element(tags::REFERENCED_SERIES_SEQUENCE, series), + ]) + }) + .collect(); + sequence_element(tag, studies) + } +} + +/// Returns the group for `key`, appending a new one if it does not exist yet. +fn group_mut<'a, T: Default>(groups: &'a mut Groups, key: &str) -> &'a mut T { + let position = groups + .iter() + .position(|(group_key, _)| group_key == key) + .unwrap_or_else(|| { + groups.push((key.to_owned(), T::default())); + groups.len() - 1 + }); + &mut groups[position].1 +} + +/// 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 reference any SOP Instance by a Referenced SOP Sequence or Referenced Study Sequence" + )] + MissingReferences, + #[error("The request payload is missing the required attribute {0}")] + MissingAttribute(Tag), + #[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::MissingReferences | Self::MissingAttribute(_) => 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/association/server.rs b/src/backend/dimse/association/server.rs index 1460084..48864c6 100644 --- a/src/backend/dimse/association/server.rs +++ b/src/backend/dimse/association/server.rs @@ -1,8 +1,10 @@ use super::{oneshot, AskPattern, Association, AssociationError, ChannelError, Command, Sender}; +use dicom::dictionary_std::uids; use dicom::transfer_syntax::TransferSyntaxRegistry; +use dicom::ul::association::server::Negotiation; use dicom::ul::association::Association as _; use dicom::ul::pdu::Pdu; -use dicom::ul::pdu::PresentationContextNegotiated; +use dicom::ul::pdu::{PresentationContextNegotiated, RequestorRoles}; use std::convert::identity; use std::io::ErrorKind; use std::{net::TcpStream, thread, time::Duration}; @@ -14,6 +16,7 @@ pub struct ServerAssociation { channel: Sender, presentation_contexts: Vec, tcp_stream: TcpStream, + peer_ae_title: String, } pub struct ServerAssociationOptions { @@ -22,12 +25,36 @@ pub struct ServerAssociationOptions { pub uncompressed: bool, } +/// Negotiates the SCP/SCU roles of the association requestor (PS3.7 D.3.3.4). +/// +/// A Storage Commitment SCP that delivers the N-EVENT-REPORT-RQ on a new association is the +/// association requestor, but must still act as the SCP of the Storage Commitment Push Model +/// SOP Class (PS3.4 J.3.3.1.2 Note 1), so it proposes the SCP role. We only accept that role, as +/// we never act as a Storage Commitment SCP ourselves. +/// For all other SOP Classes, the default roles (requestor is SCU) apply. +struct RoleNegotiation; + +impl Negotiation for RoleNegotiation { + fn negotiate_roles( + &self, + sop_class_uid: &str, + _scu_role: bool, + scp_role: bool, + ) -> Option { + (sop_class_uid == uids::STORAGE_COMMITMENT_PUSH_MODEL).then_some(RequestorRoles { + scu: false, + scp: scp_role, + }) + } +} + impl ServerAssociation { pub async fn new(options: ServerAssociationOptions) -> Result { let uuid = Uuid::new_v4(); let mut server_options = dicom::ul::ServerAssociationOptions::new() .ae_title(options.aet.clone()) - .promiscuous(true); + .promiscuous(true) + .with_negotiation(RoleNegotiation); for syntax in TransferSyntaxRegistry.iter() { if (options.uncompressed && syntax.is_codec_free()) @@ -56,13 +83,16 @@ impl ServerAssociation { ); let pcs = association.presentation_contexts().to_vec(); + let peer_ae_title = association.peer_ae_title().to_owned(); let stream = association .inner_stream() .try_clone() .expect("TcpStream::clone"); - connect_tx.send(Ok((stream, pcs))).map_err(|_value| ())?; + connect_tx + .send(Ok((stream, pcs, peer_ae_title))) + .map_err(|_value| ())?; association } Err(e) => { @@ -115,15 +145,21 @@ impl ServerAssociation { }) .map_err(AssociationError::OsThread)?; - let (tcp_stream, presentation_contexts) = + let (tcp_stream, presentation_contexts, peer_ae_title) = connect_result.await.expect("connect_result.await")?; Ok(Self { channel: tx, presentation_contexts, tcp_stream, + peer_ae_title, }) } + + /// The AE Title of the association requestor (Calling AE Title). + pub fn peer_ae_title(&self) -> &str { + &self.peer_ae_title + } } impl Association for ServerAssociation { diff --git a/src/backend/dimse/cstore/storescp.rs b/src/backend/dimse/cstore/storescp.rs index 7fac8e9..ff9e642 100644 --- a/src/backend/dimse/cstore/storescp.rs +++ b/src/backend/dimse/cstore/storescp.rs @@ -5,7 +5,9 @@ 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::{handle_event_report, 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 +29,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, }), } @@ -112,87 +121,110 @@ impl StoreServiceClassProvider { .and_then(Result::ok) .context("Missing tag COMMAND_FIELD (0000,0100)")?; + if command_field == COMMAND_FIELD_N_EVENT_REPORT_REQUEST { + handle_event_report( + &association, + association.peer_ae_title(), + message, + &inner.stgcmt_store, + ) + .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.", )); } - let message_id = message - .command - .get(tags::MESSAGE_ID) - .map(InMemElement::to_int) - .and_then(Result::ok) - .unwrap_or(0); - - let sop_class_uid = message - .command - .get(tags::AFFECTED_SOP_CLASS_UID) - .map(InMemElement::to_str) - .and_then(Result::ok) - .context("Missing tag AFFECTED_SOP_CLASS_UID (0000,0002)")?; + Self::handle_store(&association, message, &pctx.transfer_syntax, &inner).await?; + } + Ok(()) + } - let sop_instance_uid = message - .command - .get(tags::AFFECTED_SOP_INSTANCE_UID) - .map(InMemElement::to_str) - .and_then(Result::ok) - .context("Missing tag AFFECTED_SOP_INSTANCE_UID (0000,1000)")?; + /// Handles an inbound C-STORE-RQ by publishing the received instance to all subscribers + /// and responding with a C-STORE-RSP. + async fn handle_store( + association: &ServerAssociation, + message: DicomMessage, + transfer_syntax: &str, + inner: &InnerStoreServiceClassProvider, + ) -> anyhow::Result<()> { + let message_id = message + .command + .get(tags::MESSAGE_ID) + .map(InMemElement::to_int) + .and_then(Result::ok) + .unwrap_or(0); + + let sop_class_uid = message + .command + .get(tags::AFFECTED_SOP_CLASS_UID) + .map(InMemElement::to_str) + .and_then(Result::ok) + .context("Missing tag AFFECTED_SOP_CLASS_UID (0000,0002)")?; + + let sop_instance_uid = message + .command + .get(tags::AFFECTED_SOP_INSTANCE_UID) + .map(InMemElement::to_str) + .and_then(Result::ok) + .context("Missing tag AFFECTED_SOP_INSTANCE_UID (0000,1000)")?; - info!( - sop_instance_uid = sop_instance_uid.as_ref(), - sop_class_uid = sop_class_uid.as_ref(), - "Received instance" - ); - let response = CompositeStoreResponse { - sop_instance_uid: UI::from(sop_instance_uid.clone()), - sop_class_uid: UI::from(sop_class_uid.clone()), - message_id, - }; + info!( + sop_instance_uid = sop_instance_uid.as_ref(), + sop_class_uid = sop_class_uid.as_ref(), + "Received instance" + ); + let response = CompositeStoreResponse { + sop_instance_uid: UI::from(sop_instance_uid.clone()), + sop_class_uid: UI::from(sop_class_uid.clone()), + message_id, + }; - let move_originator_id = message - .command - .get(tags::MOVE_ORIGINATOR_MESSAGE_ID) - .map(InMemElement::to_int::) - .and_then(Result::ok); - - let file = message.data.unwrap().with_exact_meta( - FileMetaTableBuilder::new() - .media_storage_sop_class_uid(sop_class_uid.as_ref()) - .media_storage_sop_instance_uid(sop_instance_uid.as_ref()) - .transfer_syntax(&pctx.transfer_syntax) - .build() - .expect("FileMetaTableBuilder should contain required data"), - ); + let move_originator_id = message + .command + .get(tags::MOVE_ORIGINATOR_MESSAGE_ID) + .map(InMemElement::to_int::) + .and_then(Result::ok); + + let file = message.data.unwrap().with_exact_meta( + FileMetaTableBuilder::new() + .media_storage_sop_class_uid(sop_class_uid.as_ref()) + .media_storage_sop_instance_uid(sop_instance_uid.as_ref()) + .transfer_syntax(transfer_syntax) + .build() + .expect("FileMetaTableBuilder should contain required data"), + ); - let file = Arc::new(file); - for sub_aet in &inner.subscribers { - trace!("Publishing sub-operation result to subscriber {sub_aet}"); - let topic = SubscriptionTopic::new(AE::from(sub_aet), move_originator_id); - if let Err(err) = inner - .mediator - .publish(&topic, Ok(MoveSubOperation::Pending(Arc::clone(&file)))) - .await - { - match err { - // The receiver can drop before C-MOVE ends, e.g. after rendering the first image. - MediatorError::ChannelClosed => debug!("{err}"), - MediatorError::MissingCallback { .. } => warn!("{err}"), - } + let file = Arc::new(file); + for sub_aet in &inner.subscribers { + trace!("Publishing sub-operation result to subscriber {sub_aet}"); + let topic = SubscriptionTopic::new(AE::from(sub_aet), move_originator_id); + if let Err(err) = inner + .mediator + .publish(&topic, Ok(MoveSubOperation::Pending(Arc::clone(&file)))) + .await + { + match err { + // The receiver can drop before C-MOVE ends, e.g. after rendering the first image. + MediatorError::ChannelClosed => debug!("{err}"), + MediatorError::MissingCallback { .. } => warn!("{err}"), } } - - // Only respond after publishing the instance: The C-MOVE-SCP sends the final C-MOVE-RSP - // once all C-STORE-RSPs were received. Responding earlier would allow the C-MOVE to be - // completed before the instance was published, so the retrieve would miss the instance. - association - .write_message( - response, - message.presentation_context_id, - Duration::from_secs(10), - ) - .await?; } + + // Only respond after publishing the instance: The C-MOVE-SCP sends the final C-MOVE-RSP + // once all C-STORE-RSPs were received. Responding earlier would allow the C-MOVE to be + // completed before the instance was published, so the retrieve would miss the instance. + association + .write_message( + response, + message.presentation_context_id, + Duration::from_secs(10), + ) + .await?; Ok(()) } } diff --git a/src/backend/dimse/mod.rs b/src/backend/dimse/mod.rs index 283479f..719c6ca 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 cfind; @@ -13,6 +16,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..27ac6ac --- /dev/null +++ b/src/backend/dimse/stgcmt/mod.rs @@ -0,0 +1,314 @@ +use crate::api::stgcmt::{CommitmentResult, FailedInstance}; +use crate::api::stow::InstanceReference; +use crate::backend::dimse::association::Association; +use crate::backend::dimse::stgcmt::store::StorageCommitmentStore; +use crate::backend::dimse::{ + DicomMessage, DicomMessageWriter, 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; +use std::collections::HashMap; +use std::time::Duration; +use tracing::{info, warn}; + +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; + +/// +pub const STATUS_SUCCESS: US = 0x0000; +pub const STATUS_PROCESSING_FAILURE: US = 0x0110; + +/// Handles an inbound N-EVENT-REPORT-RQ, which reports the (asynchronous) result of a +/// previously sent Storage Commitment N-ACTION-RQ, and responds with an N-EVENT-REPORT-RSP. +/// +/// The N-EVENT-REPORT-RQ is either received by the +/// [`crate::backend::dimse::StoreServiceClassProvider`] on a new association, or on the +/// association that carried the N-ACTION-RQ, if it is still open. +/// See [`scu::StorageCommitmentServiceClassUser`]. +/// +/// The result is only recorded if it is reported by the AE the N-ACTION-RQ was sent to +/// (PS3.4 J.3.3.1.2 Note 2), given as `calling_aet`. Otherwise, the N-EVENT-REPORT-RSP indicates +/// a processing failure. +pub async fn handle_event_report( + association: &A, + calling_aet: &str, + message: DicomMessage, + store: &StorageCommitmentStore, +) -> anyhow::Result<()> { + let presentation_context_id = message.presentation_context_id; + let request = EventReportRequest::try_from(message)?; + + info!( + transaction_uid = request.transaction_uid, + calling_aet, + event_type_id = request.event_type_id, + successes = request.result.referenced_sequence.len(), + failures = request.result.failed_sequence.len(), + "Received N-EVENT-REPORT-RQ" + ); + + let status = match store.complete(&request.transaction_uid, calling_aet, request.result) { + Ok(()) => STATUS_SUCCESS, + Err(err) => { + warn!("Rejected N-EVENT-REPORT-RQ: {err}"); + STATUS_PROCESSING_FAILURE + } + }; + + let response = EventReportResponse { + message_id: request.message_id, + status, + }; + + association + .write_message(response, presentation_context_id, Duration::from_secs(10)) + .await?; + + Ok(()) +} + +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), + // Only known to the store, which holds the original request. + locations: HashMap::new(), + }; + + Ok(Self { + message_id, + event_type_id, + transaction_uid, + result, + }) + } +} + +/// N-EVENT-REPORT-RSP +pub struct EventReportResponse { + pub message_id: US, + pub status: 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, [response.status])), + 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..48749c1 --- /dev/null +++ b/src/backend/dimse/stgcmt/scu.rs @@ -0,0 +1,185 @@ +use crate::api::stow::InstanceReference; +use crate::backend::dimse::association; +use crate::backend::dimse::stgcmt::store::StorageCommitmentStore; +use crate::backend::dimse::stgcmt::{ + handle_event_report, NActionRequest, NActionResponse, COMMAND_FIELD_N_EVENT_REPORT_REQUEST, +}; +use crate::backend::dimse::{ + DicomMessageReader, DicomMessageWriter, ReadError, StatusType, WriteError, +}; +use crate::types::{AE, UI, US}; +use association::client::ClientAssociation; +use association::pool::{ + AssociationManager, AssociationPool, Object, PoolError, PresentationParameter, +}; +use association::AssociationError; +use dicom::dictionary_std::{tags, uids}; +use dicom::object::mem::InMemElement; +use std::time::Duration; +use thiserror::Error; +use tracing::{debug, trace, warn}; + +pub struct StorageCommitmentServiceClassUser { + pool: AssociationPool, + /// The AE the pool connects to, which receives the N-ACTION-RQ. + aet: AE, + timeout: Duration, + event_report_timeout: Duration, + store: StorageCommitmentStore, +} + +impl StorageCommitmentServiceClassUser { + pub const fn new( + pool: AssociationPool, + aet: AE, + timeout: Duration, + event_report_timeout: Duration, + store: StorageCommitmentStore, + ) -> Self { + Self { + pool, + aet, + timeout, + event_report_timeout, + store, + } + } + + /// 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. + pub async fn commit( + &self, + message_id: US, + transaction_uid: UI, + referenced_sop_sequence: Vec, + ) -> Result<(), CommitError> { + let mut 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, + }; + + let result = self.request(&association, request).await; + if result.is_ok() { + // The SCP may send the N-EVENT-REPORT-RQ on this association while it is still open + // (PS3.4 J.3.3), so the association must not be reused for other requests. + // If it does not, it sends the N-EVENT-REPORT-RQ on a new association instead, which + // is handled by the Store SCP. + tokio::spawn(Self::await_event_report( + association, + self.aet.clone(), + self.store.clone(), + self.event_report_timeout, + )); + } else { + // The association is left in an unknown state. + association.discard(); + } + result + } + + async fn request( + &self, + association: &ClientAssociation, + request: NActionRequest, + ) -> Result<(), CommitError> { + 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)), + } + } + + /// Waits for an N-EVENT-REPORT-RQ on the association that carried the N-ACTION-RQ, + /// then discards the association. + async fn await_event_report( + mut association: Object, + aet: AE, + store: StorageCommitmentStore, + timeout: Duration, + ) { + match association.read_message(timeout).await { + Ok(message) => { + let command_field = message + .command + .get(tags::COMMAND_FIELD) + .map(InMemElement::to_int::) + .and_then(Result::ok); + + if command_field == Some(COMMAND_FIELD_N_EVENT_REPORT_REQUEST) { + // The peer of this association is the AE the N-ACTION-RQ was sent to. + if let Err(err) = + handle_event_report(&*association, &aet, message, &store).await + { + warn!("Failed to handle N-EVENT-REPORT-RQ: {err}"); + } + } else { + warn!("Expected N-EVENT-REPORT-RQ, but received {command_field:?}"); + } + } + Err(err) => debug!("No N-EVENT-REPORT-RQ received on the N-ACTION association: {err}"), + } + + association.discard(); + } +} + +#[derive(Debug, Error)] +pub enum CommitError { + #[error(transparent)] + Read(Box), + #[error(transparent)] + Write(Box), + #[error(transparent)] + Association(#[from] PoolError), + #[error("N-ACTION-RSP indicated failure (status 0x{0:04X})")] + Failure(US), +} + +impl CommitError { + /// Whether the N-ACTION-RQ may have been received and accepted by the peer despite this + /// error, in which case the peer may still send an N-EVENT-REPORT-RQ. + pub const fn may_have_been_accepted(&self) -> bool { + match self { + // No association, or the N-ACTION-RSP explicitly rejected the request. + Self::Association(_) | Self::Failure(_) => false, + // The request could not be encoded completely, so the peer cannot have accepted it. + Self::Write(err) + if matches!(**err, WriteError::Writer(_) | WriteError::Negotiation(_)) => + { + false + } + // The request was (possibly partially) sent, but the N-ACTION-RSP was not received. + Self::Write(_) | Self::Read(_) => true, + } + } +} + +impl From for CommitError { + fn from(err: ReadError) -> Self { + Self::Read(Box::new(err)) + } +} + +impl From for CommitError { + fn from(err: WriteError) -> Self { + Self::Write(Box::new(err)) + } +} diff --git a/src/backend/dimse/stgcmt/service.rs b/src/backend/dimse/stgcmt/service.rs new file mode 100644 index 0000000..bfec001 --- /dev/null +++ b/src/backend/dimse/stgcmt/service.rs @@ -0,0 +1,72 @@ +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 crate::types::AE; +use async_trait::async_trait; +use std::time::Duration; + +pub struct DimseStgcmtService { + scu: StorageCommitmentServiceClassUser, + store: StorageCommitmentStore, + aet: AE, +} + +impl DimseStgcmtService { + pub fn new( + pool: AssociationPool, + aet: AE, + timeout: Duration, + event_report_timeout: Duration, + store: StorageCommitmentStore, + ) -> Self { + let scu = StorageCommitmentServiceClassUser::new( + pool, + aet.clone(), + timeout, + event_report_timeout, + store.clone(), + ); + Self { scu, store, aet } + } +} + +#[async_trait] +impl StgcmtService for DimseStgcmtService { + async fn commit(&self, request: CommitRequest) -> Result<(), CommitError> { + if !self.store.insert_pending( + request.transaction_uid.clone(), + self.aet.clone(), + request.locations, + ) { + return Err(CommitError::DuplicateTransaction(request.transaction_uid)); + } + + let result = self + .scu + .commit( + next_message_id(), + request.transaction_uid.clone(), + request.instances, + ) + .await; + + if let Err(err) = result { + // If the N-ACTION-RQ never reached the peer, or was explicitly rejected, no + // N-EVENT-REPORT-RQ will ever arrive for this Transaction UID - allow the client to + // retry. Otherwise (e.g. the N-ACTION-RSP timed out), the peer may still report the + // result, so the entry is kept until it arrives or expires. + if !err.may_have_been_accepted() { + self.store.remove_pending(&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..834b41b --- /dev/null +++ b/src/backend/dimse/stgcmt/store.rs @@ -0,0 +1,268 @@ +use crate::api::stgcmt::{CommitmentResult, CommitmentState, InstanceLocation}; +use crate::config::StgcmtStoreConfig; +use crate::types::{AE, UI}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use thiserror::Error; + +/// 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 after the process was restarted between the N-ACTION and the N-EVENT-REPORT is +/// rejected, as the Transaction UID is no longer known. +/// +/// Entries expire after a configurable time-to-live, and the number of entries is bounded: +/// on overflow, the least recently updated entry is dropped. +#[derive(Clone)] +pub struct StorageCommitmentStore { + inner: Arc>>, + ttl: Duration, + max_transactions: usize, +} + +struct Entry { + /// The AE Title the N-ACTION-RQ was sent to. + /// Only this AE may report the result (PS3.4 J.3.3.1.2 Note 2). + aet: AE, + updated_at: Instant, + state: State, +} + +enum State { + /// Holds the [`CommitmentResult::locations`] of the request until its result arrives. + Pending(HashMap), + Completed(CommitmentResult), +} + +/// Returned by [`StorageCommitmentStore::complete`] if the result must not be recorded. +#[derive(Debug, Error)] +pub enum CompleteError { + #[error("Unknown Transaction UID {0}")] + UnknownTransaction(UI), + #[error("Expected the result of Transaction UID {transaction_uid} from {expected}, but it was reported by {actual}")] + UnexpectedCallingAe { + transaction_uid: UI, + expected: AE, + actual: AE, + }, +} + +impl StorageCommitmentStore { + pub fn new(config: &StgcmtStoreConfig) -> Self { + Self { + inner: Arc::default(), + ttl: Duration::from_millis(config.ttl), + max_transactions: config.max_transactions, + } + } + + /// Registers a new, still-unresolved commitment request that is sent to the given AE. + /// Returns `false` if the Transaction UID is already known, per the Commit Transaction's + /// 409 (Conflict) status code. + pub fn insert_pending( + &self, + transaction_uid: UI, + aet: AE, + locations: HashMap, + ) -> bool { + let now = Instant::now(); + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + states.retain(|_, entry| now.duration_since(entry.updated_at) < self.ttl); + + if states.contains_key(&transaction_uid) { + return false; + } + + if states.len() >= self.max_transactions { + let oldest = states + .iter() + .min_by_key(|(_, entry)| entry.updated_at) + .map(|(transaction_uid, _)| transaction_uid.clone()); + if let Some(oldest) = oldest { + states.remove(&oldest); + } + } + + states.insert( + transaction_uid, + Entry { + aet, + updated_at: now, + state: State::Pending(locations), + }, + ); + true + } + + /// Records the result of a commitment request, once its N-EVENT-REPORT-RQ has arrived. + /// The result is only recorded if the Transaction UID is known and the report was sent by + /// the AE that received the N-ACTION-RQ. + pub fn complete( + &self, + transaction_uid: &str, + calling_aet: &str, + mut result: CommitmentResult, + ) -> Result<(), CompleteError> { + let now = Instant::now(); + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + let entry = states + .get_mut(transaction_uid) + .filter(|entry| now.duration_since(entry.updated_at) < self.ttl) + .ok_or_else(|| CompleteError::UnknownTransaction(transaction_uid.to_owned()))?; + + if entry.aet.trim() != calling_aet.trim() { + return Err(CompleteError::UnexpectedCallingAe { + transaction_uid: transaction_uid.to_owned(), + expected: entry.aet.clone(), + actual: calling_aet.to_owned(), + }); + } + + result.locations = match std::mem::replace(&mut entry.state, State::Pending(HashMap::new())) + { + State::Pending(locations) => locations, + // The result was delivered again, e.g. after the peer did not receive the + // N-EVENT-REPORT-RSP. + State::Completed(previous) => previous.locations, + }; + entry.state = State::Completed(result); + entry.updated_at = now; + drop(states); + Ok(()) + } + + pub fn get(&self, transaction_uid: &str) -> Option { + let now = Instant::now(); + let states = self.inner.lock().expect("mutex should not be poisoned"); + states + .get(transaction_uid) + .filter(|entry| now.duration_since(entry.updated_at) < self.ttl) + .map(|entry| match &entry.state { + State::Pending(_) => CommitmentState::Pending, + State::Completed(result) => CommitmentState::Completed(result.clone()), + }) + } + + /// 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. A completed entry is kept, as its result has already arrived. + pub fn remove_pending(&self, transaction_uid: &str) { + let mut states = self.inner.lock().expect("mutex should not be poisoned"); + if states + .get(transaction_uid) + .is_some_and(|entry| matches!(entry.state, State::Pending(_))) + { + states.remove(transaction_uid); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn store(ttl: u64, max_transactions: usize) -> StorageCommitmentStore { + StorageCommitmentStore::new(&StgcmtStoreConfig { + ttl, + max_transactions, + }) + } + + fn insert(store: &StorageCommitmentStore, transaction_uid: &str) -> bool { + store.insert_pending(transaction_uid.into(), "PACS".into(), HashMap::new()) + } + + #[test] + fn rejects_duplicate_transaction() { + let store = store(60_000, 10); + assert!(insert(&store, "1")); + assert!(!insert(&store, "1")); + } + + #[test] + fn completes_only_from_expected_ae() { + let store = store(60_000, 10); + insert(&store, "1"); + + assert!(matches!( + store.complete("1", "OTHER", CommitmentResult::default()), + Err(CompleteError::UnexpectedCallingAe { .. }) + )); + assert!(matches!(store.get("1"), Some(CommitmentState::Pending))); + + assert!(store + .complete("1", "PACS ", CommitmentResult::default()) + .is_ok()); + assert!(matches!( + store.get("1"), + Some(CommitmentState::Completed(_)) + )); + } + + #[test] + fn rejects_unknown_transaction() { + let store = store(60_000, 10); + assert!(matches!( + store.complete("1", "PACS", CommitmentResult::default()), + Err(CompleteError::UnknownTransaction(_)) + )); + assert!(store.get("1").is_none()); + } + + #[test] + fn remove_pending_keeps_completed() { + let store = store(60_000, 10); + insert(&store, "1"); + insert(&store, "2"); + store + .complete("2", "PACS", CommitmentResult::default()) + .unwrap(); + + store.remove_pending("1"); + store.remove_pending("2"); + + assert!(store.get("1").is_none()); + assert!(matches!( + store.get("2"), + Some(CommitmentState::Completed(_)) + )); + } + + #[test] + fn drops_oldest_on_overflow() { + let store = store(60_000, 2); + insert(&store, "1"); + std::thread::sleep(Duration::from_millis(1)); + insert(&store, "2"); + std::thread::sleep(Duration::from_millis(1)); + insert(&store, "3"); + + assert!(store.get("1").is_none()); + assert!(store.get("2").is_some()); + assert!(store.get("3").is_some()); + } + + #[test] + fn expires_after_ttl() { + let store = store(10, 10); + insert(&store, "1"); + std::thread::sleep(Duration::from_millis(20)); + + assert!(store.get("1").is_none()); + assert!(matches!( + store.complete("1", "PACS", CommitmentResult::default()), + Err(CompleteError::UnknownTransaction(_)) + )); + // An expired Transaction UID can be reused. + assert!(insert(&store, "1")); + } +} diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 973e37c..ba7fbb0 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,13 @@ where pool.to_owned(), Duration::from_millis(ae_config.mwl.timeout), ))), + stgcmt: Some(Box::new(DimseStgcmtService::new( + pool.to_owned(), + ae_config.aet.clone(), + Duration::from_millis(ae_config.stgcmt.timeout), + Duration::from_millis(ae_config.stgcmt.event_report_timeout), + state.stgcmt_store, + ))), } } #[cfg(feature = "s3")] @@ -88,6 +98,7 @@ where wado: Some(Box::new(S3WadoService::new(&config))), stow: None, mwl: None, + stgcmt: None, } } }; diff --git a/src/config/defaults.yaml b/src/config/defaults.yaml index 99af723..8937c40 100644 --- a/src/config/defaults.yaml +++ b/src/config/defaults.yaml @@ -15,6 +15,9 @@ server: interface: 0.0.0.0 port: 7001 uncompressed: true + stgcmt: + ttl: 86400000 + max-transactions: 10000 aets: - aet: ORTHANC host: 127.0.0.1 diff --git a/src/config/mod.rs b/src/config/mod.rs index d7047ef..8e94f1d 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,27 @@ impl Default for MwlConfig { } } +#[derive(Debug, Clone, Deserialize)] +#[serde(default, 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, + /// How long to wait for an N-EVENT-REPORT-RQ on the association that carried the + /// N-ACTION-RQ before closing it. The association is not available to other requests in the + /// meantime. Results sent later on a new association are received by the Store SCP. + pub event_report_timeout: u64, +} + +impl Default for StgcmtConfig { + fn default() -> Self { + Self { + timeout: 30_000, + event_report_timeout: 5_000, + } + } +} + impl AppConfig { /// Loads the application configuration from the following sources: /// 1. Defaults (defined in `defaults.toml`) @@ -226,6 +249,8 @@ pub struct ServerConfig { pub aet: AE, pub http: HttpServerConfig, pub dimse: Vec, + #[serde(default)] + pub stgcmt: StgcmtStoreConfig, } impl Default for ServerConfig { @@ -234,6 +259,28 @@ impl Default for ServerConfig { aet: AE::from(DEFAULT_AET), http: HttpServerConfig::default(), dimse: vec![DimseServerConfig::default()], + stgcmt: StgcmtStoreConfig::default(), + } + } +} + +/// Limits for the in-memory store of Storage Commitment transactions, shared by all AEs. +#[derive(Debug, Clone, Deserialize)] +#[serde(default, rename_all = "kebab-case")] +pub struct StgcmtStoreConfig { + /// How long (in milliseconds) a transaction is kept after its last update, i.e. after it was + /// submitted or after its result arrived. + pub ttl: u64, + /// Maximum number of transactions to keep. On overflow, the least recently updated + /// transaction is dropped. + pub max_transactions: usize, +} + +impl Default for StgcmtStoreConfig { + fn default() -> Self { + Self { + ttl: 86_400_000, // 24 h + max_transactions: 10_000, } } } diff --git a/src/main.rs b/src/main.rs index 5d9b0cc..460ad93 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; @@ -68,6 +69,7 @@ pub struct AppState { pub config: AppConfig, pub pools: AssociationPools, pub mediator: MoveMediator, + pub stgcmt_store: StorageCommitmentStore, } fn init_sentry(config: &AppConfig) -> sentry::ClientInitGuard { @@ -111,15 +113,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(&config.server.stgcmt); 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() @@ -129,7 +134,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 7e74ab9..1505ff7 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..74b8cbe --- /dev/null +++ b/tests/stgcmt.rs @@ -0,0 +1,323 @@ +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)?) +} + +/// Builds a DICOM JSON Storage Commitment Request Module (PS3.18 Annex J.1) referencing a single +/// SOP Instance by the Referenced Study Sequence instead of the Referenced SOP Sequence. +fn commit_request_body_by_study(instance: &TestInstance) -> anyhow::Result { + let sequence = |tag, item: InMemDicomObject| { + InMemElement::new( + tag, + VR::SQ, + Value::Sequence(DataSetSequence::from(vec![item])), + ) + }; + let uid = |tag, uid: &str| DataElement::new(tag, VR::UI, dicom_value!(Str, uid)); + + let sop_instance = InMemDicomObject::from_element_iter([uid( + tags::REFERENCED_SOP_INSTANCE_UID, + &instance.sop_instance_uid, + )]); + let sop_class = InMemDicomObject::from_element_iter([ + uid(tags::REFERENCED_SOP_CLASS_UID, &instance.sop_class_uid), + sequence(tags::REFERENCED_INSTANCE_SEQUENCE, sop_instance), + ]); + let series = InMemDicomObject::from_element_iter([ + uid(tags::SERIES_INSTANCE_UID, &instance.series_instance_uid), + sequence(tags::REFERENCED_INSTANCES_BY_SOP_CLASS_SEQUENCE, sop_class), + ]); + let study = InMemDicomObject::from_element_iter([ + uid(tags::STUDY_INSTANCE_UID, &instance.study_instance_uid), + sequence(tags::REFERENCED_SERIES_SEQUENCE, series), + ]); + + let mut object = InMemDicomObject::new_empty(); + object.put(sequence(tags::REFERENCED_STUDY_SEQUENCE, study)); + 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(()) +} + +struct TestInstance { + study_instance_uid: String, + series_instance_uid: String, + sop_class_uid: String, + sop_instance_uid: String, +} + +/// Storage Commitment can only be requested for instances that already exist on the origin +/// server - STOWs a test instance and returns its UIDs. +async fn stow_test_instance(http_port: u16) -> anyhow::Result { + let instance = open_file(dicom_test_files::path("pydicom/CT_small.dcm").unwrap())?; + let test_instance = TestInstance { + study_instance_uid: instance + .element(tags::STUDY_INSTANCE_UID)? + .to_str()? + .into_owned(), + series_instance_uid: instance + .element(tags::SERIES_INSTANCE_UID)? + .to_str()? + .into_owned(), + sop_class_uid: instance.meta().media_storage_sop_class_uid().to_owned(), + 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(test_instance) +} + +#[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 instance = stow_test_instance(server.http_port).await?; + let sop_class_uid = &instance.sop_class_uid; + let sop_instance_uid = &instance.sop_instance_uid; + + 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); + + // The instances may also be referenced by the Referenced Study Sequence. + let study_transaction_uid = new_transaction_uid(); + let response = http + .post(format!("{base_url}/{study_transaction_uid}")) + .header("Content-Type", "application/dicom+json") + .body(commit_request_body_by_study(&instance)?) + .send() + .await?; + assert_eq!(response.status(), 202); + + // 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" + ); + // Type 1C: only present if at least one instance failed. + assert!( + result.get(tags::FAILED_SOP_SEQUENCE).is_none(), + "Expected FailedSOPSequence to be absent" + ); + + // The response references the instances the same way as the request. + let study_result = + poll_until_completed(&http, &format!("{base_url}/{study_transaction_uid}")).await?; + assert!( + study_result.get(tags::REFERENCED_SOP_SEQUENCE).is_none() + && study_result.get(tags::FAILED_STUDY_SEQUENCE).is_none(), + "Expected only a ReferencedStudySequence" + ); + let committed_instance_uid = study_result + .get(tags::REFERENCED_STUDY_SEQUENCE) + .and_then(|element| { + element + .items()? + .first()? + .get(tags::REFERENCED_SERIES_SEQUENCE) + }) + .and_then(|element| { + element + .items()? + .first()? + .get(tags::REFERENCED_INSTANCES_BY_SOP_CLASS_SEQUENCE) + }) + .and_then(|element| { + element + .items()? + .first()? + .get(tags::REFERENCED_INSTANCE_SEQUENCE) + }) + .and_then(|element| { + element + .items()? + .first()? + .get(tags::REFERENCED_SOP_INSTANCE_UID) + }) + .map(InMemElement::to_str) + .and_then(Result::ok) + .context("Result is missing the instance in ReferencedStudySequence")?; + assert_eq!(committed_instance_uid, *sop_instance_uid); + + 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")? +}