From f9f850ffe39474574ba4cab35a86825c4798ed8f Mon Sep 17 00:00:00 2001 From: Stephan Vedder Date: Mon, 7 Sep 2026 11:42:21 +0200 Subject: [PATCH] fix: remove C-ECHO messages --- CHANGELOG.md | 11 +++++ src/backend/dimse/association/client.rs | 40 +++++++++++++++++ src/backend/dimse/association/pool.rs | 37 +++++++++++++--- src/backend/dimse/cecho/echoscu.rs | 57 ------------------------ src/backend/dimse/cecho/mod.rs | 57 ------------------------ src/backend/dimse/cfind/findscu.rs | 8 ++-- src/backend/dimse/cmove/movescu.rs | 10 ++--- src/backend/dimse/cstore/storescu.rs | 10 ++--- src/backend/dimse/mod.rs | 2 - tests/common/mod.rs | 25 ++++++++++- tests/stow.rs | 59 +++++++++++++++++++++++++ 11 files changed, 178 insertions(+), 138 deletions(-) delete mode 100644 src/backend/dimse/cecho/echoscu.rs delete mode 100644 src/backend/dimse/cecho/mod.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 14c60ee..df515a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +11,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Fixed +- Associations are now reused for subsequent requests when talking to strict service class + providers. Pooled associations were validated with a C-ECHO-RQ, but an association only + negotiates a presentation context for the abstract syntax of the actual request, so providers + that reject a C-ECHO-RQ on such a presentation context forced a new association for every + single request (e.g. for every instance of a STOW-RS request). +- Associations are no longer returned to the pool after a failed or partially completed message + exchange, as their state is unknown. + ## Changed +- Idle associations are now validated by checking the socket state instead of exchanging a + C-ECHO, saving one round trip per pooled request. + ## [0.3.0] - 2026-08-13 ### Added diff --git a/src/backend/dimse/association/client.rs b/src/backend/dimse/association/client.rs index 03f903b..8960674 100644 --- a/src/backend/dimse/association/client.rs +++ b/src/backend/dimse/association/client.rs @@ -160,6 +160,46 @@ impl ClientAssociation { pub const fn uuid(&self) -> &Uuid { &self.uuid } + + /// Cheap liveness check for pooled associations. + /// + /// This replaces the previous C-ECHO based check. A C-ECHO carries + /// `AffectedSOPClassUID = Verification (1.2.840.10008.1.1)`, but the only presentation + /// context negotiated for this association is the one of the actual request + /// (e.g. Study Root Query/Retrieve FIND). Sending the C-ECHO over that context violates + /// PS3.7. Lenient SCPs answer anyway, stricter ones reject it with a non-successful + /// status or silently drop it, which made recycling fail for every request and added the + /// full C-ECHO timeout on top. + /// + /// The socket state is sufficient to detect a dead peer: + /// - `Ok(_)`: either EOF (peer closed) or unread data from a previous message, + /// meaning the association is out of sync. Both are unusable. + /// - `WouldBlock`: nothing pending, connection still open. + /// + /// A peer that died without closing the connection is detected by the next DIMSE + /// operation, which fails and causes a new association to be established. + /// + /// Note: `O_NONBLOCK` is a property of the open file description and is therefore shared + /// with the `TcpStream` owned by the association thread. This is safe here because the + /// caller holds the association exclusively, no command is in flight, and the blocking + /// mode is restored before returning. + pub fn is_alive(&self) -> bool { + if self.channel.is_closed() { + return false; + } + + if self.tcp_stream.set_nonblocking(true).is_err() { + return false; + } + + let mut buf = [0u8; 1]; + let alive = match self.tcp_stream.peek(&mut buf) { + Ok(_) => false, + Err(err) => err.kind() == std::io::ErrorKind::WouldBlock, + }; + + self.tcp_stream.set_nonblocking(false).is_ok() && alive + } } impl Drop for ClientAssociation { diff --git a/src/backend/dimse/association/pool.rs b/src/backend/dimse/association/pool.rs index e11d40b..fa28424 100644 --- a/src/backend/dimse/association/pool.rs +++ b/src/backend/dimse/association/pool.rs @@ -1,5 +1,4 @@ use crate::backend::dimse::association; -use crate::backend::dimse::EchoServiceClassUser; use crate::config::{AppConfig, BackendConfig}; use crate::types::UI; use association::client::{ClientAssociation, ClientAssociationOptions}; @@ -130,11 +129,37 @@ pub struct Object { permit: OwnedSemaphorePermit, } +impl Object { + /// Drops the underlying object instead of returning it to the pool. + /// + /// Must be called if the object is left in an unknown state, e.g. after a failed or + /// partially completed message exchange. A liveness check cannot detect this: a message + /// that was only partially written leaves the socket quiet, so the association looks + /// reusable while the peer is still waiting for the rest of it. + /// + /// Dereferencing the object after discarding it panics. + pub fn discard(&mut self) { + self.inner = None; + } + + /// Discards this object if `result` is an error, then returns `result` unchanged. + pub fn discard_on_err(&mut self, result: Result) -> Result { + if result.is_err() { + self.discard(); + } + result + } +} + impl Deref for Object { type Target = M::Object; fn deref(&self) -> &Self::Target { - &self.inner.as_ref().unwrap().object + &self + .inner + .as_ref() + .expect("Object should not be dereferenced after being discarded") + .object } } @@ -232,11 +257,9 @@ impl Manager for AssociationManager { association } + #[allow(clippy::unused_async)] // required by the Manager trait async fn recycle(&self, association: &Self::Object) -> Result<(), String> { - let successful = EchoServiceClassUser::new(association) - .echo(Duration::from_secs(5)) - .await - .map_err(|err| format!("Failed to recycle association: {err}"))?; + let successful = association.is_alive(); if successful { info!( @@ -249,7 +272,7 @@ impl Manager for AssociationManager { backend_uuid = association.uuid().to_string(), "Recycling failed" ); - Err(String::from("C-ECHO returned non-successful status code")) + Err(String::from("Association is no longer usable")) } } } diff --git a/src/backend/dimse/cecho/echoscu.rs b/src/backend/dimse/cecho/echoscu.rs deleted file mode 100644 index 21aceeb..0000000 --- a/src/backend/dimse/cecho/echoscu.rs +++ /dev/null @@ -1,57 +0,0 @@ -use std::time::Duration; - -use thiserror::Error; -use tracing::{debug, instrument, trace}; - -use super::{CompositeEchoRequest, CompositeEchoResponse}; -use crate::backend::dimse::association; -use crate::backend::dimse::{ - next_message_id, Debug, DicomMessageReader, DicomMessageWriter, ReadError, StatusType, - WriteError, -}; -use association::client::ClientAssociation; - -/// Service class user for the Verification SOP class. -/// It simply sends a C-ECHO-RQ and waits for a C-ECHO-RSP. -/// The response contains the Status attribute that indicates the current connection status. -pub struct EchoServiceClassUser<'a> { - association: &'a ClientAssociation, -} - -impl<'a> EchoServiceClassUser<'a> { - pub const fn new(association: &'a ClientAssociation) -> Self { - Self { association } - } - - /// Initiates the C-ECHO protocol. - #[instrument(skip_all)] - pub async fn echo(&self, timeout: Duration) -> Result { - trace!("Initiated C-ECHO protocol"); - let request = CompositeEchoRequest { - message_id: next_message_id(), - }; - self.association - .write_message(request, None, timeout) - .await?; - - let response = self.association.read_message(timeout).await?; - let response = CompositeEchoResponse::try_from(response)?; - - let status_type = StatusType::try_from(response.status).unwrap_or(StatusType::Failure); - - debug!( - status = response.status, - "Received C-ECHO-RSP ({status_type:?})" - ); - Ok(status_type == StatusType::Success) - } -} - -/// Errors that can occur for the echoscu. -#[derive(Debug, Error)] -pub enum EchoError { - #[error(transparent)] - Write(#[from] WriteError), - #[error(transparent)] - Read(#[from] ReadError), -} diff --git a/src/backend/dimse/cecho/mod.rs b/src/backend/dimse/cecho/mod.rs deleted file mode 100644 index 4b34cb0..0000000 --- a/src/backend/dimse/cecho/mod.rs +++ /dev/null @@ -1,57 +0,0 @@ -mod echoscu; -pub use echoscu::*; - -use super::{DicomMessage, ReadError, DATA_SET_MISSING}; -use crate::types::US; -use dicom::core::{DataElement, VR}; -use dicom::dicom_value; -use dicom::dictionary_std::{tags, uids}; -use dicom::object::mem::InMemElement; -use dicom::object::InMemDicomObject; - -const COMMAND_FIELD_COMPOSITE_ECHO_REQUEST: US = 0x0030; - -/// C-ECHO-RQ -#[derive(Debug)] -struct CompositeEchoRequest { - message_id: US, -} - -impl From for DicomMessage { - #[rustfmt::skip] - fn from(request: CompositeEchoRequest) -> Self { - let command = InMemDicomObject::command_from_element_iter([ - DataElement::new(tags::AFFECTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::VERIFICATION)), - DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_COMPOSITE_ECHO_REQUEST])), - DataElement::new(tags::MESSAGE_ID, VR::US, dicom_value!(U16, [request.message_id])), - DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_MISSING])) - ]); - - Self { - command, - data: None, - presentation_context_id: None - } - } -} - -/// C-ECHO-RSP -#[derive(Debug)] -struct CompositeEchoResponse { - pub status: US, -} - -impl TryFrom for CompositeEchoResponse { - 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(Self::Error::MissingAttribute(tags::STATUS))?; - - Ok(Self { status }) - } -} diff --git a/src/backend/dimse/cfind/findscu.rs b/src/backend/dimse/cfind/findscu.rs index 70bb311..31e2474 100644 --- a/src/backend/dimse/cfind/findscu.rs +++ b/src/backend/dimse/cfind/findscu.rs @@ -70,13 +70,15 @@ impl FindServiceClassUser { }; try_stream! { - let association = self.pool.get(presentation).await?; + let mut association = self.pool.get(presentation).await?; let request = CompositeFindRequest::from(options); - association.write_message(request, None, self.timeout).await?; + let written = association.write_message(request, None, self.timeout).await; + association.discard_on_err(written)?; trace!("Sent C-FIND-RQ"); loop { - let response = association.read_message(self.timeout).await?; + let read = association.read_message(self.timeout).await; + let response = association.discard_on_err(read)?; let response = CompositeFindResponse::try_from(response)?; trace!("Received C-FIND-RSP"); diff --git a/src/backend/dimse/cmove/movescu.rs b/src/backend/dimse/cmove/movescu.rs index c99f611..dbac541 100644 --- a/src/backend/dimse/cmove/movescu.rs +++ b/src/backend/dimse/cmove/movescu.rs @@ -25,7 +25,7 @@ impl MoveServiceClassUser { #[instrument(skip_all, name = "MOVE-SCU")] #[allow(clippy::significant_drop_tightening)] pub async fn invoke(&self, request: CompositeMoveRequest) -> Result<(), MoveError> { - let association = self + let mut association = self .pool .get(PresentationParameter { abstract_syntax_uid: UI::from( @@ -35,13 +35,13 @@ impl MoveServiceClassUser { }) .await?; - association - .write_message(request, None, self.timeout) - .await?; + let written = association.write_message(request, None, self.timeout).await; + association.discard_on_err(written)?; trace!("Sent C-MOVE-RQ"); loop { - let response = association.read_message(self.timeout).await?; + let read = association.read_message(self.timeout).await; + let response = association.discard_on_err(read)?; trace!("Received C-MOVE-RSP"); let status_type = response diff --git a/src/backend/dimse/cstore/storescu.rs b/src/backend/dimse/cstore/storescu.rs index 88e09a6..3224f63 100644 --- a/src/backend/dimse/cstore/storescu.rs +++ b/src/backend/dimse/cstore/storescu.rs @@ -23,7 +23,7 @@ impl StoreServiceClassUser { #[allow(clippy::significant_drop_tightening)] pub async fn store(&self, file: FileDicomObject) -> Result<(), StoreError> { - let association = self + let mut association = self .pool .get(PresentationParameter { abstract_syntax_uid: UI::from(file.meta().media_storage_sop_class_uid().to_owned()), @@ -41,12 +41,12 @@ impl StoreServiceClassUser { data_set: file.into_inner(), }; - association - .write_message(request, None, self.timeout) - .await?; + let written = association.write_message(request, None, self.timeout).await; + association.discard_on_err(written)?; trace!("Sent C-STORE-RQ"); - association.read_message(self.timeout).await?; + let read = association.read_message(self.timeout).await; + association.discard_on_err(read)?; trace!("Received C-STORE-RSP"); Ok(()) diff --git a/src/backend/dimse/mod.rs b/src/backend/dimse/mod.rs index 7973e11..283479f 100644 --- a/src/backend/dimse/mod.rs +++ b/src/backend/dimse/mod.rs @@ -6,7 +6,6 @@ //! - MWL-RS is implemented as a find service class user (C-FIND service). //! -mod cecho; mod cfind; pub mod cmove; mod cstore; @@ -19,7 +18,6 @@ pub mod wado; use crate::types::{UI, US}; use association::{Association, AssociationError}; -pub use cecho::EchoServiceClassUser; pub use cstore::storescp::StoreServiceClassProvider; use dicom::dictionary_std::tags; use dicom::encoding::TransferSyntaxIndex; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f4fb392..7e74ab9 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -87,6 +87,18 @@ impl ServerProcess { .await .context("Timed out waiting for DICOM-RST to start")? } + + /// Collects log lines from the server's stdout until no new line arrives within + /// `quiet_period`. + pub async fn collect_logs(&mut self, quiet_period: Duration) -> Vec { + let mut lines = Vec::new(); + while let Ok(Ok(Some(line))) = + tokio::time::timeout(quiet_period, self.stdout.next_line()).await + { + lines.push(line); + } + lines + } } impl Drop for ServerProcess { @@ -99,6 +111,15 @@ impl Drop for ServerProcess { pub async fn with_test_environment( config: &str, test: impl AsyncFnOnce(DicomWebClient) -> anyhow::Result<()>, +) -> anyhow::Result<()> { + with_test_server(config, async |client, _server| test(client).await).await +} + +/// Like [`with_test_environment`], but also provides access to the DICOM-RST server process, +/// e.g. to inspect its log output. +pub async fn with_test_server( + config: &str, + test: impl AsyncFnOnce(DicomWebClient, &mut ServerProcess) -> anyhow::Result<()>, ) -> anyhow::Result<()> { let orthanc = spawn_orthanc().await?; let orthanc_port = orthanc @@ -107,13 +128,13 @@ pub async fn with_test_environment( .context("failed to get mapped Orthanc DIMSE port")?; let config = config.replace("${ORTHANC_PORT}", &orthanc_port.to_string()); - let server = spawn_dicomrst(&config).await?; + let mut server = spawn_dicomrst(&config).await?; let client = DicomWebClient::with_single_url(&format!( "http://localhost:{}/aets/ORTHANC", server.http_port )); - test(client).await?; + test(client, &mut server).await?; Ok(()) } diff --git a/tests/stow.rs b/tests/stow.rs index 20a6f6b..c5a65ea 100644 --- a/tests/stow.rs +++ b/tests/stow.rs @@ -171,3 +171,62 @@ async fn returns_413_if_max_upload_size_is_exceeded() -> anyhow::Result<()> { }) .await } + +// Associations must be reused for subsequent instances instead of being rebuilt for every +// single instance. Validating a pooled association must not require a DIMSE round trip, as +// the association only negotiates a presentation context for the storage SOP class of the +// instance, so a C-ECHO-RQ would be rejected by strict service class providers. +#[tokio::test] +async fn reuses_associations_for_multiple_instances() -> anyhow::Result<()> { + let config = " + 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_PORT} + backend: DIMSE + "; + + // All instances share the same SOP class and transfer syntax, so a single association is + // sufficient to store all of them. + let instances = ["pydicom/CT_small.dcm"; 3] + .map(|path| open_file(dicom_test_files::path(path).unwrap()).unwrap()); + + with_test_server(config, async |client, server| { + let response = client + .store_instances() + .with_instances(futures::stream::iter(instances)) + .run() + .await + .context("STOW-RS request failed")?; + + let referenced_sop_sequence = response + .element(tags::REFERENCED_SOP_SEQUENCE) + .context("STOW-RS response is missing ReferencedSOPSequence")?; + assert!( + referenced_sop_sequence + .items() + .is_some_and(|items| items.len() == 3), + "All three instances should appear in ReferencedSOPSequence" + ); + + let logs = server.collect_logs(Duration::from_secs(1)).await; + let created = logs + .iter() + .filter(|line| line.contains("Created new client association")) + .count(); + assert_eq!( + created, 1, + "Expected a single association for all instances, but {created} were created" + ); + + Ok(()) + }) + .await +}