h2spec ETA: 135/147

- implement reset_stream (in a very bad way) - change forcefully_terminate_stream to be quicker - more strict h2 parsing note: it bacomes apparent that: - streams lack the RFC definition of state - remultiplexing lacks a frame queue - priorizer may have to be shared among connections of a session Signed-off-by: Eloi DEMOLIS <[email protected]>
sozu-proxy · May 23, 2024 · 69241af · 69241af
1 parent 8976796
commit 69241af
Show file tree

Hide file tree

Showing 7 changed files with 348 additions and 123 deletions.
diff --git a/lib/src/protocol/mux/converter.rs b/lib/src/protocol/mux/converter.rs
@@ -1,13 +1,17 @@
 use std::str::from_utf8_unchecked;
 
-use kawa::{AsBuffer, Block, BlockConverter, Chunk, Flags, Kawa, Pair, StatusLine, Store};
-
-use crate::protocol::http::parser::compare_no_case;
+use kawa::{
+    AsBuffer, Block, BlockConverter, Chunk, Flags, Kawa, Pair, ParsingErrorKind, ParsingPhase,
+    StatusLine, Store,
+};
 
-use super::{
-    parser::{FrameHeader, FrameType, H2Error},
-    serializer::{gen_frame_header, gen_rst_stream},
-    StreamId,
+use crate::protocol::{
+    http::parser::compare_no_case,
+    mux::{
+        parser::{str_to_error_code, FrameHeader, FrameType, H2Error},
+        serializer::{gen_frame_header, gen_rst_stream},
+        StreamId,
+    },
 };
 
 pub struct H2BlockConverter<'a> {
@@ -17,6 +21,26 @@ pub struct H2BlockConverter<'a> {
 }
 
 impl<'a, T: AsBuffer> BlockConverter<T> for H2BlockConverter<'a> {
+    fn initialize(&mut self, kawa: &mut Kawa<T>) {
+        // This is very ugly... we may add a h2 variant in kawa::ParsingErrorKind
+        match kawa.parsing_phase {
+            ParsingPhase::Error {
+                kind: ParsingErrorKind::Processing { message },
+                ..
+            } => {
+                let error = str_to_error_code(message);
+                let mut frame = [0; 13];
+                gen_rst_stream(&mut frame, self.stream_id, error).unwrap();
+                kawa.push_out(Store::from_slice(&frame));
+            }
+            ParsingPhase::Error { .. } => {
+                let mut frame = [0; 13];
+                gen_rst_stream(&mut frame, self.stream_id, H2Error::InternalError).unwrap();
+                kawa.push_out(Store::from_slice(&frame));
+            }
+            _ => {}
+        }
+    }
     fn call(&mut self, block: Block, kawa: &mut Kawa<T>) {
         let buffer = kawa.storage.buffer();
         match block {
@@ -140,24 +164,18 @@ impl<'a, T: AsBuffer> BlockConverter<T> for H2BlockConverter<'a> {
                     kawa.push_out(Store::from_slice(&header));
                     kawa.push_out(Store::Alloc(payload.into_boxed_slice(), 0));
                 } else if end_stream {
-                    if kawa.is_error() {
-                        let mut frame = [0; 13];
-                        gen_rst_stream(&mut frame, self.stream_id, H2Error::InternalError).unwrap();
-                        kawa.push_out(Store::from_slice(&frame));
-                    } else {
-                        let mut header = [0; 9];
-                        gen_frame_header(
-                            &mut header,
-                            &FrameHeader {
-                                payload_len: 0,
-                                frame_type: FrameType::Data,
-                                flags: 1,
-                                stream_id: self.stream_id,
-                            },
-                        )
-                        .unwrap();
-                        kawa.push_out(Store::from_slice(&header));
-                    }
+                    let mut header = [0; 9];
+                    gen_frame_header(
+                        &mut header,
+                        &FrameHeader {
+                            payload_len: 0,
+                            frame_type: FrameType::Data,
+                            flags: 1,
+                            stream_id: self.stream_id,
+                        },
+                    )
+                    .unwrap();
+                    kawa.push_out(Store::from_slice(&header));
                 }
                 if end_header || end_stream {
                     kawa.push_delimiter()

diff --git a/lib/src/protocol/mux/h1.rs b/lib/src/protocol/mux/h1.rs
@@ -3,9 +3,7 @@ use sozu_command::ready::Ready;
 use crate::{
     println_,
     protocol::mux::{
-        debug_kawa, forcefully_terminate_answer, set_default_answer, update_readiness_after_read,
-        update_readiness_after_write, BackendStatus, Context, Endpoint, GlobalStreamId, MuxResult,
-        Position, StreamState,
+        debug_kawa, forcefully_terminate_answer, parser::H2Error, set_default_answer, update_readiness_after_read, update_readiness_after_write, BackendStatus, Context, Endpoint, GlobalStreamId, MuxResult, Position, StreamState
     },
     socket::SocketHandler,
     timer::TimeoutContainer,
@@ -226,6 +224,9 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
             Position::Client(BackendStatus::Connected(cluster_id))
             | Position::Client(BackendStatus::Connecting(cluster_id)) => {
                 self.stream = usize::MAX;
+                // keep alive should probably be used only if the http context is fully reset
+                // in case end_stream occurs due to an error the connection state is probably
+                // unrecoverable and should be terminated
                 if stream_context.keep_alive_backend {
                     self.position =
                         Position::Client(BackendStatus::KeepAlive(std::mem::take(cluster_id)))
@@ -241,7 +242,7 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
                     // if the answer is not terminated we send an RstStream to properly clean the stream
                     // if it is terminated, we finish the transfer, the backend is not necessary anymore
                     if !stream.back.is_terminated() {
-                        forcefully_terminate_answer(stream, &mut self.readiness);
+                        forcefully_terminate_answer(stream, &mut self.readiness, H2Error::InternalError);
                     } else {
                         stream.state = StreamState::Unlinked;
                         self.readiness.interest.insert(Ready::WRITABLE);