fix(body): validate body sender lifecycle

lib/wreq_ruby/body.rb

@@ -2,10 +2,10 @@
 
 unless defined?(Wreq)
   module Wreq
-    # BodySender: for streaming request bodies, allowing thread-safe chunked data push.
-    # Backed by a Rust channel, avoids buffering the entire payload in memory.
+    # Streams a request body through a bounded, thread-safe channel.
     #
-    # Supports multi-threaded chunk push: you can safely call `push` from multiple threads.
+    # The channel applies backpressure instead of buffering the entire request body.
+    # Multiple threads may safely push chunks while a request drains the receiving side.
     #
     # Usage:
     #   sender = Wreq::BodySender.new(8)
@@ -15,25 +15,43 @@ module Wreq
     #   end
     #   resp = client.post(url, body: sender)
     #
-    # Note:
-    #   - Sender is for request upload only, not for response reading.
-    #   - Each BodySender instance can only be used once (single-use):
-    #     after being consumed by a request, further push or reuse is not allowed.
+    # A sender can be attached to one request. Closing it prevents further writes but
+    # retains queued chunks so a request attached afterward can still drain them.
     class BodySender
-      # @param capacity [Integer] channel buffer size, default 8
+      # Create a bounded request-body sender.
+      #
+      # @param capacity [Integer] positive number of chunks that may wait in the channel;
+      #   defaults to 8 and must be greater than zero
       # @return [Wreq::BodySender] A streaming request body sender
+      # @raise [ArgumentError] if capacity is zero, negative, or too large
+      # @raise [TypeError] if capacity is not an Integer
       def self.new(capacity = 8)
       end
 
+      # Push one binary chunk, waiting while the channel is full.
+      #
       # @param data [String] binary chunk
-      # @return [void]
+      # @return [nil]
+      # @raise [IOError] if the sender or receiving side is closed
       def push(data)
       end
 
-      # Close the sender, signaling end of data.
-      # @return [void]
+      # Close the producer and signal EOF after all queued chunks are read.
+      #
+      # This operation is idempotent.
+      #
+      # @return [nil]
       def close
       end
+
+      # Return whether the sender can no longer accept chunks.
+      #
+      # This becomes true after {#close} or when the request stops consuming
+      # the receiving side.
+      #
+      # @return [Boolean]
+      def closed?
+      end
     end
   end
 end

src/client/body.rs

@@ -47,5 +47,6 @@ pub fn include(ruby: &Ruby, gem_module: &RModule) -> Result<(), Error> {
     sender_class.define_singleton_method("new", function!(stream::BodySender::new, -1))?;
     sender_class.define_method("push", method!(stream::BodySender::push, 1))?;
     sender_class.define_method("close", magnus::method!(stream::BodySender::close, 0))?;
+    sender_class.define_method("closed?", magnus::method!(stream::BodySender::is_closed, 0))?;
     Ok(())
 }

src/client/body/stream.rs

@@ -1,36 +1,59 @@
 //! Streaming request and response body support.
 
 use std::{
+    cell::{Ref, RefCell, RefMut},
+    panic::catch_unwind,
     pin::Pin,
-    sync::RwLock,
     task::{Context, Poll},
 };
 
 use bytes::Bytes;
 use futures_util::{Stream, StreamExt};
-use magnus::{Error, RString, TryConvert, Value};
-use tokio::sync::{
-    Mutex,
-    mpsc::{self},
-};
+use magnus::{Error, Integer, RString, Ruby, Value};
+use tokio::sync::{Mutex, Semaphore, mpsc};
 
 use crate::{
-    error::{memory_error, mpsc_send_error_to_magnus, wreq_error_to_magnus},
+    error::{
+        body_sender_borrow_error_to_magnus, body_sender_borrow_mut_error_to_magnus,
+        body_sender_send_error_to_magnus, closed_body_sender_error, memory_error,
+        wreq_error_to_magnus,
+    },
     rt,
 };
 
+/// Number of chunks buffered when Ruby omits the channel capacity.
+const DEFAULT_CHANNEL_CAPACITY: usize = 8;
+
 /// A receiver for streaming HTTP response bodies.
 pub struct BodyReceiver(Mutex<Pin<Box<dyn Stream<Item = wreq::Result<Bytes>> + Send>>>);
 
-/// A sender for streaming HTTP request bodies.
+/// A bounded producer for a single streaming HTTP request body.
+///
+/// The receiving side may be attached to one request. The producer remains
+/// writable until [`BodySender::close`] is called or the request drops its
+/// receiver. Ruby's GVL protects state access; no [`RefCell`] borrow is kept
+/// while request backpressure waits without the GVL.
 #[magnus::wrap(class = "Wreq::BodySender", free_immediately, size)]
-pub struct BodySender(RwLock<InnerBodySender>);
+pub struct BodySender(RefCell<InnerBodySender>);
 
+/// Mutable ownership state for both halves of the body channel.
 struct InnerBodySender {
+    /// Producing side, removed by [`BodySender::close`].
     tx: Option<mpsc::Sender<Bytes>>,
+    /// Receiving side, removed when the sender is attached to a request.
     rx: Option<mpsc::Receiver<Bytes>>,
 }
 
+impl InnerBodySender {
+    /// Return whether the channel can no longer accept body chunks.
+    fn is_closed(&self) -> bool {
+        match &self.tx {
+            Some(tx) => tx.is_closed(),
+            None => true,
+        }
+    }
+}
+
 // ===== impl BodyReceiver =====
 
 impl BodyReceiver {
@@ -58,47 +81,136 @@ impl BodyReceiver {
 // ===== impl BodySender =====
 
 impl BodySender {
-    /// Ruby: `Wreq::Sender.new(capacity = 8)`
-    pub fn new(args: &[Value]) -> Self {
-        let capacity: usize = if let Some(v) = args.first() {
-            usize::try_convert(*v).unwrap_or(8)
-        } else {
-            8
-        };
+    /// Create a bounded request-body channel.
+    ///
+    /// Ruby: `Wreq::BodySender.new(capacity = 8)`. Capacity must be greater
+    /// than zero and no larger than [`Semaphore::MAX_PERMITS`].
+    ///
+    /// # Errors
+    ///
+    /// Returns `TypeError` for a non-Integer capacity and `ArgumentError` for
+    /// an invalid range or argument count.
+    pub fn new(ruby: &Ruby, args: &[Value]) -> Result<Self, Error> {
+        let capacity = parse_capacity(ruby, args)?;
+
+        // Create the Tokio channel without allowing an unwind to cross the Ruby FFI boundary.
+        //
+        // Known panic conditions are rejected by [`parse_capacity`]. The unwind guard
+        // remains as a defensive fallback if Tokio adds another channel invariant.
+        let (tx, rx) =
+            catch_unwind(|| mpsc::channel(capacity)).map_err(|_| invalid_capacity_error(ruby))?;
 
-        let (tx, rx) = mpsc::channel(capacity);
-        BodySender(RwLock::new(InnerBodySender {
+        Ok(BodySender(RefCell::new(InnerBodySender {
             tx: Some(tx),
             rx: Some(rx),
-        }))
+        })))
     }
 
-    /// Ruby: `push(data)` where data is String or bytes
+    /// Push a binary chunk, waiting for capacity when the channel is full.
+    ///
+    /// Ruby: `push(data)` where `data` is a String.
+    ///
+    /// # Errors
+    ///
+    /// Returns `IOError` after either channel side has closed. An interrupted
+    /// wait retains the existing `Wreq::InterruptError` behavior.
     pub fn push(rb_self: &Self, data: RString) -> Result<(), Error> {
-        let bytes = data.to_bytes();
-        let inner = rb_self.0.read().unwrap();
-        if let Some(ref tx) = inner.tx {
-            rt::try_block_on(tx.send(bytes), mpsc_send_error_to_magnus)?;
-        }
-        Ok(())
+        // Clone during the shared borrow, then release it before waiting
+        // for capacity. Request attachment needs a mutable borrow.
+        let tx = match &rb_self.read_inner()?.tx {
+            Some(tx) if !tx.is_closed() => tx.clone(),
+            _ => return Err(closed_body_sender_error()),
+        };
+
+        rt::try_block_on(tx.send(data.to_bytes()), body_sender_send_error_to_magnus)
     }
 
-    /// Ruby: `close` to close the sender
-    pub fn close(&self) {
-        let mut inner = self.0.write().unwrap();
+    /// Close the producing side while retaining the receiver and queued chunks.
+    ///
+    /// Calling this method more than once has no additional effect.
+    ///
+    /// # Errors
+    ///
+    /// Returns `Wreq::BodyError` if the internal state is already borrowed.
+    pub fn close(&self) -> Result<(), Error> {
+        let mut inner = self.write_inner()?;
         inner.tx.take();
-        inner.rx.take();
+        Ok(())
+    }
+
+    /// Return whether this sender can no longer accept body chunks.
+    ///
+    /// # Errors
+    ///
+    /// Returns `Wreq::BodyError` if the internal state is already borrowed.
+    pub fn is_closed(&self) -> Result<bool, Error> {
+        self.read_inner().map(|r| r.is_closed())
+    }
+
+    /// Borrow the channel state without panicking on accidental re-entry.
+    fn read_inner(&self) -> Result<Ref<'_, InnerBodySender>, Error> {
+        self.0
+            .try_borrow()
+            .map_err(body_sender_borrow_error_to_magnus)
     }
+
+    /// Mutably borrow the channel state without panicking on accidental re-entry.
+    fn write_inner(&self) -> Result<RefMut<'_, InnerBodySender>, Error> {
+        self.0
+            .try_borrow_mut()
+            .map_err(body_sender_borrow_mut_error_to_magnus)
+    }
+}
+
+/// Parse and validate the optional Ruby channel capacity.
+///
+/// [`mpsc::channel`] panics for zero or values above
+/// [`Semaphore::MAX_PERMITS`], so validation must finish before channel creation.
+fn parse_capacity(ruby: &Ruby, args: &[Value]) -> Result<usize, Error> {
+    let value = match args {
+        [] => return Ok(DEFAULT_CHANNEL_CAPACITY),
+        [value] => *value,
+        _ => {
+            return Err(Error::new(
+                ruby.exception_arg_error(),
+                format!(
+                    "wrong number of arguments (given {}, expected 0..1)",
+                    args.len()
+                ),
+            ));
+        }
+    };
+
+    let integer = Integer::from_value(value)
+        .ok_or_else(|| Error::new(ruby.exception_type_error(), "capacity must be an Integer"))?;
+    let capacity = integer
+        .to_i64()
+        .ok()
+        .and_then(|capacity| usize::try_from(capacity).ok())
+        .filter(|capacity| (1..=Semaphore::MAX_PERMITS).contains(capacity))
+        .ok_or_else(|| invalid_capacity_error(ruby))?;
+
+    Ok(capacity)
+}
+
+/// Build the synchronous Ruby error used for an invalid channel capacity.
+fn invalid_capacity_error(ruby: &Ruby) -> Error {
+    Error::new(
+        ruby.exception_arg_error(),
+        format!("capacity must be between 1 and {}", Semaphore::MAX_PERMITS),
+    )
 }
 
+/// Move the receiving side into one request body.
+///
+/// The sender remains available for concurrent producers until it is closed or
+/// the returned stream is dropped. A second attachment returns `Wreq::MemoryError`.
 impl TryFrom<&BodySender> for ReceiverStream<Bytes> {
     type Error = magnus::Error;
 
     fn try_from(sender: &BodySender) -> Result<Self, Self::Error> {
         sender
-            .0
-            .write()
-            .unwrap()
+            .write_inner()?
             .rx
             .take()
             .map(ReceiverStream::new)

src/error.rs

@@ -1,3 +1,5 @@
+use std::cell::{BorrowError, BorrowMutError};
+
 use magnus::{
     Error as MagnusError, RModule, Ruby, exception::ExceptionClass, prelude::*, value::Lazy,
 };
@@ -90,11 +92,32 @@ pub fn no_block_given_error() -> MagnusError {
     )
 }
 
-/// Map [`tokio::sync::mpsc::error::SendError`] to corresponding [`magnus::Error`]
-pub fn mpsc_send_error_to_magnus<T>(err: SendError<T>) -> MagnusError {
+/// Build an `IOError` for writes to a closed request-body sender.
+pub fn closed_body_sender_error() -> MagnusError {
+    MagnusError::new(ruby!().exception_io_error(), "closed body sender")
+}
+
+/// Map a failed body-channel send to `IOError`.
+pub fn body_sender_send_error_to_magnus<T>(err: SendError<T>) -> MagnusError {
+    MagnusError::new(
+        ruby!().exception_io_error(),
+        format!("closed body sender: {err}"),
+    )
+}
+
+/// Map an immutable sender-state borrow failure to `Wreq::BodyError`.
+pub fn body_sender_borrow_error_to_magnus(err: BorrowError) -> MagnusError {
+    MagnusError::new(
+        ruby!().get_inner(&BODY_ERROR),
+        format!("body sender state is unavailable: {err}"),
+    )
+}
+
+/// Map a mutable sender-state borrow failure to `Wreq::BodyError`.
+pub fn body_sender_borrow_mut_error_to_magnus(err: BorrowMutError) -> MagnusError {
     MagnusError::new(
         ruby!().get_inner(&BODY_ERROR),
-        format!("failed to send body chunk: {}", err),
+        format!("body sender state is unavailable: {err}"),
     )
 }