sqlx-postgres: Cleanup and add more comments

Author: joeydewaal

sqlx-core/src/net/socket/buffered.rs

@@ -70,7 +70,7 @@ impl<S: Socket> BufferedSocket<S> {
         F: FnMut(&mut BytesMut) -> Result<ControlFlow<R, usize>, Error>,
     {
         loop {
-            // Read if we want bytes
+            // Ensure we have enough bytes, only read if we want bytes.
             ready!(self.poll_handle_read(cx)?);
 
             match try_read(&mut self.read_buf.read)? {
@@ -149,7 +149,7 @@ impl<S: Socket> BufferedSocket<S> {
     pub async fn flush(&mut self) -> io::Result<()> {
         while !self.write_buf.is_empty() {
             let written = self.socket.write(self.write_buf.get()).await?;
-            // Consume does the sanity check
+            // Consume does the sanity check.
             self.write_buf.consume(written);
         }
 
@@ -402,7 +402,7 @@ impl<S: Socket> Sink<&[u8]> for BufferedSocket<S> {
 
         while !this.write_buf.is_empty() {
             let written = ready!(this.socket.poll_write(cx, this.write_buf.get())?);
-            // Consume does the sanity check
+            // Consume does the sanity check.
             this.write_buf.consume(written);
         }
         this.socket.poll_flush(cx).map_err(Into::into)

sqlx-postgres/src/connection/establish.rs

@@ -49,6 +49,8 @@ impl PgConnection {
             params.push(("options", options));
         }
 
+        // Only after establishing a connection, Postgres sends a [ReadyForQuery] response. While
+        // establishing a connection this pipe is used to read responses from.
         let mut pipe = conn.pipe(|buf| {
             buf.write(Startup {
                 username: Some(&options.username),

sqlx-postgres/src/connection/worker.rs

@@ -95,7 +95,7 @@ impl Worker {
             Poll::Ready(Some(request)) => Poll::Ready(request),
             Poll::Ready(None) => {
                 // Channel was closed, explicitly or because the sender was dropped. Either way
-                // we should start a gracefull shutdown.
+                // we should start a graceful shutdown.
                 self.socket
                     .write_buffer_mut()
                     .put_slice(&[Terminate::FORMAT as u8, 0, 0, 0, 4]);
@@ -151,6 +151,8 @@ impl Worker {
         while let Poll::Ready(response) = self.poll_next_message(cx)? {
             match response.format {
                 BackendMessageFormat::ReadyForQuery => {
+                    // Cloning a `ReceivedMessage` here is cheap because it only clones the
+                    // underlying `Bytes`
                     let rfq: ReadyForQuery = response.clone().decode()?;
                     self.shared.set_transaction_status(rfq.transaction_status);
 
@@ -187,6 +189,7 @@ impl Worker {
         }
 
         if self.state != WorkerState::Open && self.back_log.is_empty() {
+            // After the connection is closed and the backlog is empty we can close the socket.
             self.state = WorkerState::Closed;
         }
         Ok(())
@@ -232,18 +235,15 @@ impl Worker {
     #[inline(always)]
     fn poll_shutdown(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
         if self.state == WorkerState::Closed {
-            // The buffer is closed, a [Terminate] message has been sent, now try and close the socket.
+            // The channel is closed, all requests are flushed and a [Terminate] message has been
+            // sent, now try and close the socket
             self.socket.poll_close_unpin(cx)
         } else {
             Poll::Pending
         }
     }
-}
-
-impl Future for Worker {
-    type Output = Result<()>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll_worker(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
         // Try to receive responses from the database and handle them.
         self.poll_backlog(cx)?;
 
@@ -260,6 +260,17 @@ impl Future for Worker {
     }
 }
 
+impl Future for Worker {
+    type Output = Result<()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.poll_worker(cx).map_err(|e| {
+            tracing::error!("Background worker stopped with error: {e:?}");
+            e
+        })
+    }
+}
+
 #[derive(Clone)]
 pub struct Shared(Arc<Mutex<SharedInner>>);