feat(ipc): Implement Backpressure Detection with Unit Tests (#68)

* feat(ipc): Implement backpressure detection and warnings

This commit introduces a backpressure detection mechanism across all IPC transport types to provide better diagnostics during long-running or high-throughput benchmarks.

When a transport's underlying buffer becomes full, send operations can block or slow down, impacting performance metrics. This feature detects such scenarios and issues a one-time warning to the user, helping them understand if their benchmark is measuring pure transport speed or a backpressure-limited scenario.

Implementation Details:

- The IpcTransport::send method signature has been changed to return a Result<bool>, where true indicates that backpressure was detected during the send operation.

- For Shared Memory, backpressure is detected when the ring buffer is full, causing the send operation to retry.

- For TCP and Unix Domain Sockets, backpressure is detected heuristically by timing the send call. An unusually long duration suggests the OS send buffer is full and the call was blocked.

- For POSIX Message Queues, backpressure is detected when a send operation returns an EAGAIN error, indicating a full queue.

- The benchmark runner has been updated to handle the new send signature.

AI-assisted-by: Gemini 2.5 Pro

feat(ipc): Add unit tests for backpressure detection

This commit introduces unit tests for the backpressure detection
feature across all IPC transport types. These tests validate that
backpressure is correctly identified when buffers or queues become full.

Key changes:
- Added `test_shared_memory_backpressure` to verify backpressure
  detection in the shared memory transport.
- Added `test_tcp_socket_backpressure` to test the heuristic-based
  backpressure detection in TCP sockets.
- Added `test_uds_backpressure` for Unix Domain Sockets, which uses a
  similar heuristic to TCP.
- Added `test_pmq_backpressure` to ensure backpressure is detected
  when the POSIX Message Queue is full.

Additionally, the core logic for stream-based transports (TCP and UDS)
has been improved to use timeouts on send operations. This prevents
tests from hanging and provides a more robust backpressure signal. A
custom `IpcError::BackpressureTimeout` is now returned when a send
operation times out due to a full buffer, making the behavior
consistent across all transport types.

AI-assisted-by: Gemini 2.5 Pro

docs(ipc): Document backpressure detection feature

This commit
  updates the documentation to reflect the new backpressure detection and warning
  mechanism.

Key changes include:
   - Updated doc comments for the \IpcTransport::send\ method and the new
     \IpcError::BackpressureTimeout\ variant to explain the behavior.
   - Added doc comments to each transport's \send\ implementation detailing its specific
     backpressure detection strategy.
   - Added a "Backpressure Warnings" section to the \README.md\ to help users interpret
     benchmark results correctly.
   - Updated \CONTRIBUTING.md\ to require that new transport implementations include
     backpressure detection logic.

AI-assisted-by: Gemini 2.5 Pro

* remove errant added AGENTS.md file

Author: dustinblack

CONTRIBUTING.md

@@ -120,6 +120,15 @@ Use descriptive branch names:
 - `docs/improve-readme` - Documentation
 - `perf/optimize-serialization` - Performance improvements
 
+### Implementing a New Transport
+
+When adding a new IPC transport mechanism, it must implement the `IpcTransport` trait defined in `src/ipc/mod.rs`. Key requirements include:
+
+- **Asynchronous Operations**: All I/O methods must be `async`.
+- **Bidirectional Support**: The transport should ideally support sending and receiving.
+- **Backpressure Detection**: The `send` method is required to detect when the transport's buffer is full and the operation is blocked. It must return `Ok(true)` when backpressure is detected. See the existing implementations for examples.
+- **Resource Cleanup**: The `close` method and the `Drop` trait must be implemented to ensure all system resources (files, memory segments, etc.) are properly cleaned up.
+
 ## Code Style
 
 ### Rust Style Guidelines

Cargo.lock

@@ -40,6 +40,7 @@ socket2 = "0.5"
 tracing-appender = "0.2.3"
 time = ">=0.3.34, <0.3.36" # Pinned to a compatible range for MSRV 1.70
 parking_lot = "0.12.4"
+thiserror = "2.0.16"
 
 [dev-dependencies]
 criterion = "0.5"

Cargo.toml

@@ -346,6 +346,20 @@ taskset -c 0-3 ipc-benchmark --concurrency 4
 - **Noise**: Run tests on idle systems for best accuracy
 - **Repetition**: Run multiple test executions and average results
 
+### Backpressure Warnings
+
+During high-throughput tests, it's possible for the sender to produce data faster than the receiver can consume it. When the underlying OS buffers or message queues become full, the send operation will slow down or block. This condition is known as **backpressure**.
+
+This benchmark tool automatically detects backpressure and will issue a warning message the first time it occurs for a given transport, for example:
+
+```
+WARN rusty_comms::ipc::shared_memory: Shared memory buffer is full; backpressure is occurring. 
+This may impact latency and throughput measurements. 
+Consider increasing the buffer size if this is not the desired scenario.
+```
+
+When you see this warning, it means your benchmark may be measuring a backpressure-limited scenario rather than the pure, unconstrained performance of the IPC mechanism. This is a valid and important scenario to test, but it's crucial for interpreting the results correctly. If your goal is to measure maximum throughput, you may need to increase the buffer sizes (`--buffer-size`) or investigate the receiver's performance.
+
 ## Troubleshooting
 
 ### Common Issues

README.md

@@ -431,7 +431,7 @@ impl BenchmarkRunner {
         let payload = vec![0u8; self.config.message_size];
         for i in 0..self.config.warmup_iterations {
             let message = Message::new(i as u64, payload.clone(), MessageType::OneWay);
-            client_transport.send(&message).await?;
+            let _ = client_transport.send(&message).await?;
         }
 
         // Clean up resources
@@ -682,7 +682,7 @@ impl BenchmarkRunner {
                 )
                 .await
                 {
-                    Ok(Ok(())) => {
+                    Ok(Ok(_)) => {
                         let latency = send_time.elapsed();
                         metrics_collector
                             .record_message(self.config.message_size, Some(latency))?;
@@ -716,7 +716,7 @@ impl BenchmarkRunner {
             for i in 0..msg_count {
                 let send_time = Instant::now();
                 let message = Message::new(i as u64, payload.clone(), MessageType::OneWay);
-                client_transport.send(&message).await?;
+                let _ = client_transport.send(&message).await?;
 
                 let latency = send_time.elapsed();
                 metrics_collector.record_message(self.config.message_size, Some(latency))?;
@@ -859,7 +859,7 @@ impl BenchmarkRunner {
                 )
                 .await
                 {
-                    Ok(Ok(())) => {
+                    Ok(Ok(_)) => {
                         // Send succeeded - increment counter immediately to ensure unique message IDs
                         i += 1;
 
@@ -911,7 +911,7 @@ impl BenchmarkRunner {
             for i in 0..msg_count {
                 let send_time = Instant::now();
                 let message = Message::new(i as u64, payload.clone(), MessageType::Request);
-                client_transport.send(&message).await?;
+                let _ = client_transport.send(&message).await?;
 
                 let _ = client_transport.receive().await?;
                 let latency = send_time.elapsed();
@@ -1231,7 +1231,7 @@ impl BenchmarkRunner {
                 )
                 .await
                 {
-                    Ok(Ok(())) => {
+                    Ok(Ok(_)) => {
                         let one_way_latency = send_start.elapsed();
 
                         // Try to receive response and measure round-trip latency
@@ -1292,7 +1292,7 @@ impl BenchmarkRunner {
                 let send_start = Instant::now();
                 let message = Message::new(i as u64, payload.clone(), MessageType::Request);
 
-                client_transport.send(&message).await?;
+                let _ = client_transport.send(&message).await?;
                 let one_way_latency = send_start.elapsed();
 
                 let _ = client_transport.receive().await?;

src/benchmark.rs

@@ -49,6 +49,7 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
+use thiserror::Error;
 use time::OffsetDateTime;
 use tokio::sync::mpsc;
 
@@ -68,6 +69,21 @@ pub use tcp_socket::TcpSocketTransport;
 #[cfg(unix)]
 pub use unix_domain_socket::UnixDomainSocketTransport;
 
+/// Custom error types for IPC operations.
+#[derive(Error, Debug)]
+pub enum IpcError {
+    /// Error indicating a timeout that occurred due to backpressure.
+    ///
+    /// This error is returned when a send operation cannot complete within a
+    /// reasonable time because the underlying buffer or queue is full,
+    /// indicating that the receiver is not processing messages fast enough.
+    #[error("Timeout sending message due to backpressure")]
+    BackpressureTimeout,
+    /// A generic IPC error.
+    #[error("IPC error: {0}")]
+    Generic(#[from] anyhow::Error),
+}
+
 /// Connection identifier for tracking multiple client connections
 ///
 /// This type alias provides a clear identifier for individual connections
@@ -471,15 +487,15 @@ pub trait IpcTransport: Send + Sync {
     /// - `message`: Message to transmit
     ///
     /// ## Returns
-    /// - `Ok(())`: Message sent successfully
+    /// - `Ok(bool)`: `true` if backpressure was detected, `false` otherwise
     /// - `Err(anyhow::Error)`: Transmission failed
     ///
     /// ## Performance Considerations
     ///
     /// This method should be optimized for low latency and high throughput
     /// as it's called frequently during benchmarking. Implementations
     /// should minimize copies and allocations where possible.
-    async fn send(&mut self, message: &Message) -> Result<()>;
+    async fn send(&mut self, message: &Message) -> Result<bool>;
 
     /// Receive a message (legacy single-connection interface)
     ///
@@ -644,7 +660,8 @@ pub trait IpcTransport: Send + Sync {
         message: &Message,
     ) -> Result<()> {
         // Default implementation ignores connection_id and uses legacy send
-        self.send(message).await
+        self.send(message).await?;
+        Ok(())
     }
 
     /// Get list of active connection IDs