Repository: bearcove/loona (language: rust)

Issue:
h2: Have process_frame return H2Error, do `send_goaway` from outside of it
Instead of calling `send_goaway` manually everywhere.


Discussion / hints:

Fix patch:
diff --git a/crates/fluke/src/h2/read.rs b/crates/fluke/src/h2/read.rs
index 4a43fae2..ed8f019c 100644
--- a/crates/fluke/src/h2/read.rs
+++ b/crates/fluke/src/h2/read.rs
@@ -126,14 +126,27 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                 payload_roll
             };
 
-            self.process_frame(frame, payload).await?;
+            if let Err(e) = self.process_frame(frame, payload).await {
+                match e {
+                    H2Error::Connection(e) => {
+                        self.send_goaway(e).await;
+                        // TODO: keep track that we sent a goaway.
+                        // TODO: don't let the inner functions send a goaway themselves,
+                        // so all goaways have to go through here?
+                        // keep looping
+                    }
+                    H2Error::Stream(id, e) => {
+                        self.send_rst(id, e).await;
+                    }
+                }
+            }
         }
     }
 
     // TODO: return `H2Error` from this, which can accommodate both connection-level
     // and stream-level errors.
     // cf.
-    async fn process_frame(&mut self, frame: Frame, payload: Roll) -> eyre::Result<()> {
+    async fn process_frame(&mut self, frame: Frame, payload: Roll) -> H2Result {
         match self.continuation_state {
             ContinuationState::Idle => self.process_idle_frame(frame, payload).await,
             ContinuationState::ContinuingHeaders(expected_stream_id) => {
@@ -144,7 +157,7 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
         }
     }
 
-    async fn process_idle_frame(&mut self, frame: Frame, mut payload: Roll) -> eyre::Result<()> {
+    async fn process_idle_frame(&mut self, frame: Frame, mut payload: Roll) -> H2Result {
         match frame.frame_type {
             FrameType::Data(flags) => {
                 if flags.contains(DataFlags::Padded) {
@@ -190,11 +203,9 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                             tx
                         }
                         StreamStage::HalfClosedRemote | StreamStage::Closed => {
-                            let err = H2ConnectionError::ReceivedDataForClosedStream {
-                                stream_id: frame.stream_id,
-                            };
-                            self.send_goaway(err).await;
-                            return Ok(());
+                            return H2StreamError::ReceivedDataForClosedStream
+                                .for_stream(frame.stream_id)
+                                .into();
                         }
                     }
                 };
@@ -212,21 +223,17 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                 // than the last one we accepted.
 
                 if frame.stream_id.is_server_initiated() {
-                    self.send_goaway(H2ConnectionError::ClientSidShouldBeOdd)
-                        .await;
-                    return Ok(());
+                    return H2ConnectionError::ClientSidShouldBeOdd.into();
                 }
 
                 if frame.stream_id < self.state.last_stream_id {
-                    self.send_goaway(H2ConnectionError::ClientSidShouldBeIncreasing)
-                        .await;
+                    return H2ConnectionError::ClientSidShouldBeIncreasing.into();
                 }
                 self.state.last_stream_id = frame.stream_id;
 
                 let padding_length = if flags.contains(HeadersFlags::Padded) {
                     if payload.is_empty() {
-                        self.send_goaway(H2ConnectionError::PaddedFrameEmpty).await;
-                        return Ok(());
+                        return H2ConnectionError::PaddedFrameEmpty.into();
                     }
 
                     let padding_length_roll;
@@ -314,7 +321,7 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                 }
 
                 if flags.contains(HeadersFlags::EndHeaders) {
-                    self.end_headers(frame.stream_id).await;
+                    self.end_headers(frame.stream_id).await?;
                 } else {
                     debug!(
                         "expecting more headers/trailers for stream {}",
@@ -361,7 +368,9 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                         .await
                         .is_err()
                     {
-                        return Err(eyre::eyre!("could not send H2 acknowledge settings event"));
+                        return Err(
+                            eyre::eyre!("could not send H2 acknowledge settings event").into()
+                        );
                     }
                 }
             }
@@ -387,7 +396,7 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                 }
 
                 if self.ev_tx.send(H2ConnEvent::Ping(payload)).await.is_err() {
-                    return Err(eyre::eyre!("could not send H2 ping event"));
+                    return Err(eyre::eyre!("could not send H2 ping event").into());
                 }
             }
             FrameType::GoAway => todo!(),
@@ -430,26 +439,24 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
         frame: Frame,
         payload: Roll,
         expected_stream_id: StreamId,
-    ) -> eyre::Result<()> {
+    ) -> H2Result {
         let flags = match frame.frame_type {
             FrameType::Continuation(flags) => flags,
             other => {
-                self.send_goaway(H2ConnectionError::ExpectedContinuationFrame {
+                return H2ConnectionError::ExpectedContinuationFrame {
                     stream_id: expected_stream_id,
                     frame_type: other,
-                })
-                .await;
-                return Ok(());
+                }
+                .into()
             }
         };
 
         if frame.stream_id != expected_stream_id {
-            self.send_goaway(H2ConnectionError::ExpectedContinuationForStream {
+            return H2ConnectionError::ExpectedContinuationForStream {
                 stream_id: expected_stream_id,
                 continuation_stream_id: frame.stream_id,
-            })
-            .await;
-            return Ok(());
+            }
+            .into();
         }
 
         // unwrap rationale: we just checked that this is a
@@ -468,14 +475,14 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
         }
 
         if flags.contains(ContinuationFlags::EndHeaders) {
-            self.end_headers(frame.stream_id).await;
+            self.end_headers(frame.stream_id).await
         } else {
             debug!(
                 "expecting more headers/trailers for stream {}",
                 frame.stream_id
             );
+            Ok(())
         }
-        Ok(())
     }
 
     async fn send_goaway(&self, err: H2ConnectionError) {
@@ -497,9 +504,8 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
     }
 
     async fn send_rst(&mut self, stream_id: StreamId, e: H2StreamError) {
-        warn!("stream error: {e:?}");
         let error_code = e.as_known_error_code();
-        debug!("error_code = {error_code:?}");
+        warn!("sending rst because: {e} (known error code: {error_code:?})");
 
         if self
             .ev_tx
@@ -512,26 +518,15 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
         {
             debug!("error sending rst");
         }
-    }
 
-    async fn end_headers(&mut self, stream_id: StreamId) {
-        if let Err(err) = self.end_headers_inner(stream_id).await {
-            match err {
-                H2Error::Connection(e) => {
-                    self.send_goaway(e).await;
-                }
-                H2Error::Stream(e) => {
-                    self.send_rst(stream_id, e).await;
-                }
-            }
-        }
+        self.state.streams.remove(&stream_id);
     }
 
     // TODO: this can all be, again, greatly simplified because continuation
     // frames are supposed to follow the headers frames directly, so we don't
     // need any of that in the state machine, we can just keep popping more
     // frames.
-    async fn end_headers_inner(&mut self, stream_id: StreamId) -> H2Result {
+    async fn end_headers(&mut self, stream_id: StreamId) -> H2Result {
         let stage = self
             .state
             .streams
diff --git a/crates/fluke/src/h2/types.rs b/crates/fluke/src/h2/types.rs
index f3fdc09e..25988af9 100644
--- a/crates/fluke/src/h2/types.rs
+++ b/crates/fluke/src/h2/types.rs
@@ -148,11 +148,20 @@ pub(crate) enum H2ConnectionError {
     #[error("max concurrent streams exceeded (more than {max_concurrent_streams})")]
     MaxConcurrentStreamsExceeded { max_concurrent_streams: u32 },
 
-    #[error("received data for closed stream {stream_id}")]
-    ReceivedDataForClosedStream { stream_id: StreamId },
-
     #[error("other error: {0:?}")]
-    Other(#[from] eyre::Report),
+    Internal(#[from] eyre::Report),
+}
+
+impl<T> From<H2ConnectionError> for H2Result<T> {
+    fn from(e: H2ConnectionError) -> Self {
+        Err(e.into())
+    }
+}
+
+impl<T> From<H2Error> for H2Result<T> {
+    fn from(e: H2Error) -> Self {
+        Err(e)
+    }
 }
 
 impl H2ConnectionError {
@@ -164,10 +173,8 @@ impl H2ConnectionError {
             H2ConnectionError::PaddedFrameTooShort => KnownErrorCode::FrameSizeError,
             // compression errors
             H2ConnectionError::CompressionError(_) => KnownErrorCode::CompressionError,
-            // stream closed errors
-            H2ConnectionError::ReceivedDataForClosedStream { .. } => KnownErrorCode::StreamClosed,
             // internal errors
-            H2ConnectionError::Other(_) => KnownErrorCode::InternalError,
+            H2ConnectionError::Internal(_) => KnownErrorCode::InternalError,
             // protocol errors
             _ => KnownErrorCode::ProtocolError,
         }
@@ -181,19 +188,27 @@ pub(crate) enum H2Error {
     #[error("connection error: {0:?}")]
     Connection(#[from] H2ConnectionError),
 
-    #[error("stream error: {0:?}")]
-    Stream(#[from] H2StreamError),
+    #[error("stream error: for stream {0}: {1:?}")]
+    Stream(StreamId, H2StreamError),
+}
+
+impl From<eyre::Report> for H2Error {
+    fn from(e: eyre::Report) -> Self {
+        Self::Connection(H2ConnectionError::Internal(e))
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum H2StreamError {
     #[allow(dead_code)]
-    #[error("for stream {stream_id}, received {data_length} bytes in data frames but content-length announced {content_length} bytes")]
+    #[error("received {data_length} bytes in data frames but content-length announced {content_length} bytes")]
     DataLengthDoesNotMatchContentLength {
-        stream_id: StreamId,
         data_length: u64,
         content_length: u64,
     },
+
+    #[error("received data for closed stream")]
+    ReceivedDataForClosedStream,
 }
 
 impl H2StreamError {
@@ -203,8 +218,13 @@ impl H2StreamError {
 
         match self {
             DataLengthDoesNotMatchContentLength { .. } => Code::ProtocolError,
+            ReceivedDataForClosedStream => Code::StreamClosed,
         }
     }
+
+    pub(crate) fn for_stream(self, stream_id: StreamId) -> H2Error {
+        H2Error::Stream(stream_id, self)
+    }
 }
 
 #[derive(Debug, Clone, Copy)]


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 913a89f4..7855c6dc 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -17,30 +17,29 @@ jobs:
       RUSTC_WRAPPER: "sccache"
     steps:
       - name: Check out repository code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           fetch-depth: 2
       - name: Install Rust specified toolchain
         run: rustup show
+      - uses: Swatinem/rust-cache@v2
+        with:
+          cache-targets: false # sccache does that
+          cache-all-crates: true
+          cache-on-failure: true
       - name: Run sccache-cache
         uses: mozilla-actions/sccache-action@v0.0.3
-      - name: Install cargo-nextest
-        uses: taiki-e/install-action@nextest
       - name: Run cargo check
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --all-targets
+        run: |
+          cargo check --all-targets
       - name: Build testbed
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --release --manifest-path test-crates/hyper-testbed/Cargo.toml
+        run: |
+          cargo build --release --manifest-path test-crates/hyper-testbed/Cargo.toml
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@nextest
       - name: Run cargo nextest
-        uses: actions-rs/cargo@v1
-        with:
-          command: nextest
-          args: run --profile ci
+        run: |
+          cargo nextest run --profile ci
   test-windows:
     runs-on: windows-2022
     env:
@@ -50,30 +49,29 @@ jobs:
       RUSTC_WRAPPER: "sccache"
     steps:
       - name: Check out repository code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           fetch-depth: 2
       - name: Install Rust specified toolchain
         run: rustup show
+      - uses: Swatinem/rust-cache@v2
+        with:
+          cache-targets: false # sccache does that
+          cache-all-crates: true
+          cache-on-failure: true
       - name: Run sccache-cache
         uses: mozilla-actions/sccache-action@v0.0.3
-      - name: Install cargo-nextest
-        uses: taiki-e/install-action@nextest
       - name: Run cargo check
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --all-targets
+        run: |
+          cargo check --all-targets
       - name: Build testbed
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --release --manifest-path test-crates/hyper-testbed/Cargo.toml
+        run: |
+          cargo build --release --manifest-path test-crates/hyper-testbed/Cargo.toml
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@nextest
       - name: Run cargo nextest
-        uses: actions-rs/cargo@v1
-        with:
-          command: nextest
-          args: run --profile ci
+        run: |
+          cargo nextest run --profile ci
   test-linux:
     env:
       CARGO_TERM_COLOR: always
@@ -83,7 +81,7 @@ jobs:
     runs-on: self-hosted
     steps:
       - name: Check out repository code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           fetch-depth: 2
       - name: Run unit tests and h2spec