do more clean shutdown

.github/workflows/codecov.yml

@@ -63,15 +63,16 @@ jobs:
     runs-on: ubuntu-24.04
     services:
       ftp-server:
-        image: ghcr.io/garethflowers/ftp-server:edge
+        image: bogem/ftp:latest
         env:
           FTP_USER: ftpuser
           FTP_PASS: ftppass
+          PASV_ADDRESS: 127.0.0.1
         ports:
           - '20-21:20-21/tcp'
-          - '40000-40009:40000-40009/tcp'
+          - '47400-47470:47400-47470/tcp'
         volumes:
-          - /tmp/ftp:/home/user
+          - /tmp/vsftpd:/home/vsftpd
       httpbin:
         image: ghcr.io/psf/httpbin:0.10.2
         ports:

g3proxy/src/inspect/http/v1/mod.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use async_recursion::async_recursion;
 use http::{Method, Version};
 use slog::slog_info;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 
 use g3_dpi::Protocol;
 use g3_io_ext::{FlexBufReader, LimitedBufReadExt};
@@ -150,7 +150,11 @@ where
 
                 r = req_acceptor.accept() => match r {
                     Some(r) => r,
-                    None => return Ok(None),
+                    None => {
+                        let _ = rsp_io.ups_w.shutdown().await;
+                        let _ = rsp_io.clt_w.shutdown().await;
+                        return Ok(None);
+                    }
                 },
                 r = rsp_io.ups_r.fill_wait_data() => {
                     req_acceptor.close();
@@ -164,7 +168,10 @@ where
 
             self.req_id += 1;
             match r {
-                HttpRecvRequest::ClientConnectionClosed => return Ok(None),
+                HttpRecvRequest::ClientConnectionClosed => {
+                    let _ = rsp_io.ups_w.shutdown().await;
+                    return Ok(None);
+                }
                 HttpRecvRequest::ClientConnectionError(e) => return Err(e),
                 HttpRecvRequest::ClientRequestError(e) => {
                     if let Some(rsp) =

g3proxy/src/inspect/imap/mod.rs

@@ -393,10 +393,14 @@ where
             CloseReason::Client => {
                 self.handle_client_logout(&mut clt_w, &mut ups_r, &mut relay_buf.rsp_recv_buf)
                     .await?;
+                let _ = ups_w.shutdown().await;
+                let _ = clt_w.shutdown().await;
                 Ok(())
             }
             CloseReason::Server => {
                 self.mark_close_by_server();
+                let _ = ups_w.shutdown().await;
+                let _ = clt_w.shutdown().await;
                 Ok(())
             }
             CloseReason::Local(e) => {

g3proxy/src/inspect/smtp/mod.rs

@@ -342,7 +342,11 @@ where
                 )
                 .await?;
             match next_action {
-                ForwardNextAction::Quit => return Ok(None),
+                ForwardNextAction::Quit => {
+                    let _ = ups_w.shutdown().await;
+                    let _ = clt_w.shutdown().await;
+                    return Ok(None);
+                }
                 ForwardNextAction::StartTls => {
                     return if let Some(tls_interception) = self.ctx.tls_interception() {
                         let mut start_tls_obj =
@@ -393,6 +397,8 @@ where
                         )
                         .await?;
                     if transaction.quit() {
+                        let _ = ups_w.shutdown().await;
+                        let _ = clt_w.shutdown().await;
                         return Ok(None);
                     }
                 }

g3proxy/src/module/http_forward/connection/eof_poller.rs

@@ -40,16 +40,12 @@ impl HttpConnectionEofCheck {
             _ = conn.1.fill_wait_data() => {
                 // close early when EOF or unexpected data, to avoid waiting at other side
                 wait_channel.close();
-                // make sure we correctly shutdown tls connection
-                // FIXME use async drop at escaper side when supported
                 let _ = conn.0.shutdown().await;
             }
             v = &mut wait_channel => {
                 if matches!(v, Ok(true)) {
                     let _ = send_channel.send(conn);
                 } else {
-                    // make sure we correctly shutdown tls connection
-                    // FIXME use async drop at escaper side when supported
                     let _ = conn.0.shutdown().await;
                 }
             }