Merge pull request #1097 from sozu-proxy/devel/edemolis/fix/tls-flush

Expose internal rustls buffers to ensure they are flushed
sozu-proxy · Apr 3, 2024 · f306a79 · f306a79
2 parents c17d173 + d83f399
commit f306a79
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 13 deletions.
diff --git a/lib/src/protocol/kawa_h1/editor.rs b/lib/src/protocol/kawa_h1/editor.rs
@@ -95,9 +95,9 @@ impl HttpContext {
                 .map(ToOwned::to_owned);
         }
 
-        if self.method == Some(Method::Get) && request.body_size == kawa::BodySize::Empty {
-            request.parsing_phase = kawa::ParsingPhase::Terminated;
-        }
+        // if self.method == Some(Method::Get) && request.body_size == kawa::BodySize::Empty {
+        //     request.parsing_phase = kawa::ParsingPhase::Terminated;
+        // }
 
         let public_ip = self.public_address.ip();
         let public_port = self.public_address.port();

diff --git a/lib/src/protocol/kawa_h1/mod.rs b/lib/src/protocol/kawa_h1/mod.rs
@@ -433,7 +433,7 @@ impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> Http<Front, L
         self.response_stream.prepare(&mut kawa::h1::BlockConverter);
 
         let bufs = self.response_stream.as_io_slice();
-        if bufs.is_empty() {
+        if bufs.is_empty() && !self.frontend_socket.socket_wants_write() {
             self.frontend_readiness.interest.remove(Ready::WRITABLE);
             return StateResult::Continue;
         }
@@ -452,8 +452,6 @@ impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> Http<Front, L
             count!("bytes_out", size as i64);
             metrics.bout += size;
             self.backend_readiness.interest.insert(Ready::READABLE);
-        } else {
-            self.frontend_readiness.event.remove(Ready::WRITABLE);
         }
 
         match socket_state {
@@ -475,6 +473,10 @@ impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> Http<Front, L
             SocketResult::Continue => {}
         }
 
+        if self.frontend_socket.socket_wants_write() {
+            return StateResult::Continue;
+        }
+
         if self.response_stream.is_terminated() && self.response_stream.is_completed() {
             self.log_request_success(metrics);
             metrics.reset();

diff --git a/lib/src/socket.rs b/lib/src/socket.rs
@@ -48,8 +48,9 @@ pub enum TransportProtocol {
 pub trait SocketHandler {
     fn socket_read(&mut self, buf: &mut [u8]) -> (usize, SocketResult);
     fn socket_write(&mut self, buf: &[u8]) -> (usize, SocketResult);
-    fn socket_write_vectored(&mut self, _buf: &[std::io::IoSlice]) -> (usize, SocketResult) {
-        unimplemented!()
+    fn socket_write_vectored(&mut self, _buf: &[std::io::IoSlice]) -> (usize, SocketResult);
+    fn socket_wants_write(&self) -> bool {
+        false
     }
     fn socket_ref(&self) -> &TcpStream;
     fn socket_mut(&mut self) -> &mut TcpStream;
@@ -183,7 +184,7 @@ impl SocketHandler for FrontRustls {
             counter += 1;
             if counter > MAX_LOOP_ITERATIONS {
                 error!("MAX_LOOP_ITERATION reached in FrontRustls::socket_read");
-                incr!("socket.read.infinite_loop.error");
+                incr!("rustls.read.infinite_loop.error");
             }
 
             if size == buf.len() {
@@ -232,7 +233,6 @@ impl SocketHandler for FrontRustls {
                     Ok(0) => break,
                     Ok(sz) => {
                         size += sz;
-                        can_read = true;
                     }
                     Err(e) => match e.kind() {
                         ErrorKind::WouldBlock => {
@@ -276,7 +276,7 @@ impl SocketHandler for FrontRustls {
             counter += 1;
             if counter > MAX_LOOP_ITERATIONS {
                 error!("MAX_LOOP_ITERATION reached in FrontRustls::socket_write");
-                incr!("socket.write.infinite_loop.error");
+                incr!("rustls.write.infinite_loop.error");
             }
             if buffered_size == buf.len() {
                 break;
@@ -392,11 +392,10 @@ impl SocketHandler for FrontRustls {
             counter += 1;
             if counter > MAX_LOOP_ITERATIONS {
                 error!("MAX_LOOP_ITERATION reached in FrontRustls::socket_write_vectored");
-                incr!("socket.write.infinite_loop.error");
+                incr!("rustls.write.infinite_loop.error");
             }
             match self.session.write_tls(&mut self.stream) {
                 Ok(0) => {
-                    //can_write = false;
                     break;
                 }
                 Ok(_sz) => {}
@@ -433,6 +432,10 @@ impl SocketHandler for FrontRustls {
         }
     }
 
+    fn socket_wants_write(&self) -> bool {
+        self.session.wants_write()
+    }
+
     fn socket_ref(&self) -> &TcpStream {
         &self.stream
     }