feat(docs): Write etl crate Rust docs (#262)

Author: iambriccardo

etl/src/concurrency/shutdown.rs

@@ -2,36 +2,69 @@ use tokio::sync::watch;
 
 use crate::concurrency::signal::{SignalRx, SignalTx, create_signal};
 
+/// Transmitter side of the shutdown coordination channel.
+///
+/// [`ShutdownTx`] enables sending shutdown signals to multiple workers simultaneously.
+/// It wraps a signal transmitter with shutdown-specific semantics and provides methods
+/// for triggering shutdown and creating receiver subscriptions.
 #[derive(Debug, Clone)]
 pub struct ShutdownTx(SignalTx);
 
 impl ShutdownTx {
+    /// Wraps a signal transmitter with shutdown semantics.
     pub fn wrap(tx: SignalTx) -> Self {
         Self(tx)
     }
 
+    /// Triggers shutdown for all subscribed workers.
+    ///
+    /// This method broadcasts a shutdown signal to all workers that have subscribed
+    /// to this shutdown channel. Workers should respond by completing their current
+    /// operations gracefully and terminating.
     pub fn shutdown(&self) -> Result<(), watch::error::SendError<()>> {
         self.0.send(())
     }
 
+    /// Creates a new shutdown receiver for worker subscription.
+    ///
+    /// Each worker should call this method to get its own receiver that can be used
+    /// to detect when shutdown has been requested. Multiple receivers can be created
+    /// from the same transmitter.
     pub fn subscribe(&self) -> ShutdownRx {
         self.0.subscribe()
     }
 }
 
+/// Receiver side of the shutdown coordination channel.
+///
+/// [`ShutdownRx`] is used by workers to detect when shutdown has been requested.
+/// It's a type alias for [`SignalRx`] with shutdown-specific semantics.
 pub type ShutdownRx = SignalRx;
 
+/// Result type that distinguishes between normal operation and shutdown scenarios.
+///
+/// [`ShutdownResult`] is used by operations that can be interrupted by shutdown signals.
+/// It preserves both successful results and any partial data that was being processed
+/// when shutdown was requested.
 pub enum ShutdownResult<T, I> {
+    /// Normal successful completion with result data.
     Ok(T),
+    /// Operation was interrupted by shutdown, with any partial data preserved.
     Shutdown(I),
 }
 
 impl<T, I> ShutdownResult<T, I> {
+    /// Returns true if this result represents a shutdown scenario.
     pub fn should_shutdown(&self) -> bool {
         matches!(self, ShutdownResult::Shutdown(_))
     }
 }
 
+/// Creates a new shutdown coordination channel.
+///
+/// This function creates a broadcast channel for coordinating shutdown across multiple
+/// workers. The transmitter can be used to trigger shutdown, while receivers can be
+/// distributed to workers that need to respond to shutdown signals.
 pub fn create_shutdown_channel() -> (ShutdownTx, ShutdownRx) {
     let (tx, rx) = create_signal();
     (ShutdownTx::wrap(tx), rx)

etl/src/concurrency/signal.rs

@@ -1,12 +1,24 @@
 use tokio::sync::watch;
 
-/// Type alias to abstract a watch channel of `()`.
+/// Transmitter side of a coordination signal channel.
+///
+/// [`SignalTx`] abstracts a watch channel transmitter for sending coordination signals
+/// between workers. The signal carries no data payload - it's purely for notification
+/// that some event or state change has occurred.
 pub type SignalTx = watch::Sender<()>;
 
-/// Type alias to abstract a watch channel of `()`.
+/// Receiver side of a coordination signal channel.
+///
+/// [`SignalRx`] abstracts a watch channel receiver for detecting coordination signals.
+/// Workers can use this to wait for events from other parts of the system without
+/// polling or complex synchronization.
 pub type SignalRx = watch::Receiver<()>;
 
-/// Creates a new pair of [`SignalTx`] and [`SignalRx`].
+/// Creates a new coordination signal channel.
+///
+/// This function creates a watch-based signaling channel optimized for coordination
+/// scenarios where multiple receivers need to be notified of the same event. Unlike
+/// mpsc channels, all receivers see the same signal simultaneously.
 pub fn create_signal() -> (SignalTx, SignalRx) {
     let (tx, rx) = watch::channel(());
     (tx, rx)

etl/src/concurrency/stream.rs

@@ -53,48 +53,45 @@ impl<B, S: Stream<Item = B>> BatchStream<B, S> {
 impl<B, S: Stream<Item = B>> Stream for BatchStream<B, S> {
     type Item = ShutdownResult<Vec<S::Item>, Vec<S::Item>>;
 
-    /// Polls the stream for the next batch of items.
+    /// Polls the stream for the next batch of items using a complex state machine.
     ///
-    /// Returns:
-    /// - `Poll::Ready(Some(batch))` when a complete batch is available
-    /// - `Poll::Ready(None)` when the stream has ended
-    /// - `Poll::Pending` when more items are needed to form a batch
-    ///
-    /// The stream will emit a batch when:
-    /// - The batch reaches maximum size
-    /// - A timeout occurs
-    /// - The stream is forcefully stopped
+    /// This method implements a batching algorithm that balances throughput
+    /// and latency by collecting items into batches based on both size and time constraints.
+    /// The polling state machine handles multiple concurrent conditions and ensures proper
+    /// resource cleanup during shutdown scenarios.
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.as_mut().project();
 
+        // Fast path: if the inner stream has already ended, we're done
         if *this.inner_stream_ended {
             return Poll::Ready(None);
         }
 
         loop {
+            // Fast path: if we've been marked as stopped, terminate immediately
             if *this.stream_stopped {
                 return Poll::Ready(None);
             }
 
-            // If the stream has been asked to stop, we mark the stream as stopped and return the
-            // remaining elements, irrespectively of boundaries.
+            // PRIORITY 1: Check for shutdown signal
+            // Shutdown handling takes priority over all other operations to ensure
+            // graceful termination. We return any accumulated items with shutdown indication.
             if this.shutdown_rx.has_changed().unwrap_or(false) {
                 info!("the stream has been forcefully stopped");
 
-                // We mark the stream as stopped, in this way any further call with return
-                // `Poll::Ready(None)`.
+                // Mark stream as permanently stopped to prevent further polling
                 *this.stream_stopped = true;
 
-                // We mark the current value as unchanged, effectively acknowledging that we have
-                // seen it. This does not affect the correctness, but it makes the implementation
-                // semantically more correct.
+                // Acknowledge that we've seen the shutdown signal to maintain watch semantics
                 this.shutdown_rx.mark_unchanged();
 
-                // Even if we have no items, we return this result, since we signal that a shutdown
-                // signal was received and the consumer side of the stream, can decide what to do.
+                // Return accumulated items (if any) with shutdown indication
+                // Even empty batches are returned to signal shutdown occurred
                 return Poll::Ready(Some(ShutdownResult::Shutdown(std::mem::take(this.items))));
             }
 
+            // PRIORITY 2: Timer management
+            // Reset the timeout timer when starting a new batch or after emitting a batch
             if *this.reset_timer {
                 this.deadline
                     .set(Some(tokio::time::sleep(Duration::from_millis(
@@ -103,48 +100,63 @@ impl<B, S: Stream<Item = B>> Stream for BatchStream<B, S> {
                 *this.reset_timer = false;
             }
 
+            // PRIORITY 3: Memory optimization
+            // Pre-allocate batch capacity when starting to collect items
+            // This avoids reallocations during batch collection
             if this.items.is_empty() {
                 this.items.reserve_exact(this.batch_config.max_size);
             }
 
+            // PRIORITY 4: Poll underlying stream for new items
             match this.stream.as_mut().poll_next(cx) {
-                Poll::Pending => break,
+                Poll::Pending => {
+                    // No more items available right now, check if we should emit due to timeout
+                    break;
+                }
                 Poll::Ready(Some(item)) => {
+                    // New item available - add to current batch
                     this.items.push(item);
 
-                    // If we reached the `max_batch_size` we want to return the batch and reset the
-                    // timer.
+                    // SIZE-BASED EMISSION: If batch is full, emit immediately
+                    // This provides throughput optimization for high-volume streams
                     if this.items.len() >= this.batch_config.max_size {
-                        *this.reset_timer = true;
+                        *this.reset_timer = true; // Schedule timer reset for next batch
                         return Poll::Ready(Some(ShutdownResult::Ok(std::mem::take(this.items))));
                     }
+                    // Continue loop to collect more items or check other conditions
                 }
                 Poll::Ready(None) => {
+                    // STREAM END: Underlying stream finished
+                    // Return final batch if we have items, otherwise signal completion
                     let last = if this.items.is_empty() {
-                        None
+                        None // No final batch needed
                     } else {
-                        *this.reset_timer = true;
+                        *this.reset_timer = true; // Clean up timer state
                         Some(ShutdownResult::Ok(std::mem::take(this.items)))
                     };
 
-                    *this.inner_stream_ended = true;
+                    *this.inner_stream_ended = true; // Mark stream as permanently ended
 
                     return Poll::Ready(last);
                 }
             }
         }
 
-        // If there are items, we want to check the deadline, if it's met, we return the batch
-        // we currently have in memory, otherwise, we return.
+        // PRIORITY 5: Time-based emission check
+        // If we have items and the timeout has expired, emit the current batch
+        // This provides latency bounds to prevent indefinite delays in low-volume scenarios
         if !this.items.is_empty()
             && let Some(deadline) = this.deadline.as_pin_mut()
         {
+            // Check if timeout has elapsed (this will register waker if not ready)
             ready!(deadline.poll(cx));
-            *this.reset_timer = true;
+
+            *this.reset_timer = true; // Schedule timer reset for next batch
 
             return Poll::Ready(Some(ShutdownResult::Ok(std::mem::take(this.items))));
         }
 
+        // No conditions met for batch emission - wait for more items or timeout
         Poll::Pending
     }
 }

etl/src/config/mod.rs

@@ -1,2 +1,6 @@
+//! Configuration objects for ETL pipelines.
+//!
+//! Re-exports configuration types and utilities required for pipeline setup and operation.
+
 // Re-exports.
 pub use etl_config::shared::*;

etl/src/conversions/bool.rs

@@ -2,6 +2,11 @@ use crate::bail;
 use crate::error::EtlResult;
 use crate::error::{ErrorKind, EtlError};
 
+/// Parses a PostgreSQL boolean value from its text format representation.
+///
+/// PostgreSQL represents boolean values in text format as single characters:
+/// - `"t"` → `true` (exactly one lowercase 't')
+/// - `"f"` → `false` (exactly one lowercase 'f')
 pub fn parse_bool(s: &str) -> EtlResult<bool> {
     if s == "t" {
         Ok(true)