fix: remove C-ECHO messages

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

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 {

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<M: Manager> {
 	permit: OwnedSemaphorePermit,
 }
 
+impl<M: Manager> Object<M> {
+	/// 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<T, E>(&mut self, result: Result<T, E>) -> Result<T, E> {
+		if result.is_err() {
+			self.discard();
+		}
+		result
+	}
+}
+
 impl<M: Manager> Deref for Object<M> {
 	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"))
 		}
 	}
 }

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");
 

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

src/backend/dimse/cstore/storescu.rs

@@ -23,7 +23,7 @@ impl StoreServiceClassUser {
 
 	#[allow(clippy::significant_drop_tightening)]
 	pub async fn store(&self, file: FileDicomObject<InMemDicomObject>) -> 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(())

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;

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<String> {
+		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(())
 }