Repository: bearcove/loona (language: rust)

Issue:
h2: Don't have a `Headers` state, just pop frames until we're done reading header continuations
This will simplify GREATLY parsing headers

edit: we can kill the Trailers state too, omg.


Discussion / hints:

Fix patch:
diff --git a/Justfile b/Justfile
index 3e1281c1..b2e5076c 100644
--- a/Justfile
+++ b/Justfile
@@ -20,6 +20,9 @@ test *args:
 	just build-testbed
 	RUST_BACKTRACE=1 cargo nextest run {{args}}
 
+curl-tests:
+	RUST_BACKTRACE=1 cargo nextest run --manifest-path test-crates/fluke-curl-tests/Cargo.toml
+
 build-testbed:
 	cargo build --release --manifest-path test-crates/hyper-testbed/Cargo.toml
 	
@@ -32,7 +35,7 @@ bench *args:
 h2spec *args:
 	#!/bin/bash -eux	
 	export RUST_LOG="${RUST_LOG:-fluke=debug,fluke_hpack=info}"
-	export RUST_BACKTRACE=1
+	export RUST_BACKTRACE="${RUST_BACKTRACE:-1}"
 	cargo run --manifest-path test-crates/fluke-h2spec/Cargo.toml -- {{args}}
 
 check:
diff --git a/crates/fluke-buffet/src/piece.rs b/crates/fluke-buffet/src/piece.rs
index d590a4c2..e69deb50 100644
--- a/crates/fluke-buffet/src/piece.rs
+++ b/crates/fluke-buffet/src/piece.rs
@@ -2,8 +2,8 @@
 
 use std::{fmt, ops::Deref, str::Utf8Error};
 
-use http::header::HeaderName;
 use fluke_maybe_uring::buf::IoBuf;
+use http::header::HeaderName;
 
 use crate::{Roll, RollStr};
 
diff --git a/crates/fluke/src/h2/body.rs b/crates/fluke/src/h2/body.rs
index 7c22ebd3..adbafc91 100644
--- a/crates/fluke/src/h2/body.rs
+++ b/crates/fluke/src/h2/body.rs
@@ -9,6 +9,7 @@ pub(crate) enum PieceOrTrailers {
 }
 
 pub(crate) type H2BodySender = mpsc::Sender<H2BodyItem>;
+// FIXME: don't use eyre, do proper error handling
 pub(crate) type H2BodyItem = eyre::Result<PieceOrTrailers>;
 
 #[derive(Debug)]
diff --git a/crates/fluke/src/h2/parse.rs b/crates/fluke/src/h2/parse.rs
index 1916393e..832cb5d8 100644
--- a/crates/fluke/src/h2/parse.rs
+++ b/crates/fluke/src/h2/parse.rs
@@ -274,19 +274,23 @@ impl Frame {
 }
 
 /// See https://httpwg.org/specs/rfc9113.html#FrameHeader - the first bit
-/// is reserved, and the rest is a 32-bit stream id
-fn parse_reserved_and_stream_id(i: Roll) -> IResult<Roll, (u8, StreamId)> {
+/// is reserved, and the rest is a 31-bit stream id
+pub fn parse_reserved_and_u31(i: Roll) -> IResult<Roll, (u8, u32)> {
     fn reserved(i: (Roll, usize)) -> IResult<(Roll, usize), u8> {
         nom::bits::streaming::take(1_usize)(i)
     }
 
-    fn stream_id(i: (Roll, usize)) -> IResult<(Roll, usize), StreamId> {
-        nom::combinator::map(nom::bits::streaming::take(31_usize), StreamId)(i)
+    fn stream_id(i: (Roll, usize)) -> IResult<(Roll, usize), u32> {
+        nom::bits::streaming::take(31_usize)(i)
     }
 
     nom::bits::bits(tuple((reserved, stream_id)))(i)
 }
 
+fn parse_reserved_and_stream_id(i: Roll) -> IResult<Roll, (u8, StreamId)> {
+    parse_reserved_and_u31(i).map(|(i, (reserved, stream_id))| (i, (reserved, StreamId(stream_id))))
+}
+
 // cf. https://httpwg.org/specs/rfc9113.html#HEADERS
 #[derive(Debug)]
 pub(crate) struct PrioritySpec {
diff --git a/crates/fluke/src/h2/read.rs b/crates/fluke/src/h2/read.rs
index 277bb4cd..e59c9971 100644
--- a/crates/fluke/src/h2/read.rs
+++ b/crates/fluke/src/h2/read.rs
@@ -1,5 +1,11 @@
-use std::{borrow::Cow, rc::Rc};
+use std::{
+    borrow::Cow,
+    rc::Rc,
+    sync::atomic::{AtomicU32, Ordering},
+};
 
+use enumflags2::BitFlags;
+use eyre::Context;
 use fluke_buffet::{Piece, PieceStr, Roll, RollMut};
 use fluke_maybe_uring::io::ReadOwned;
 use http::{
@@ -8,14 +14,14 @@ use http::{
     HeaderName, Version,
 };
 use nom::Finish;
-use smallvec::smallvec;
+use smallvec::{smallvec, SmallVec};
 use tokio::sync::mpsc;
 use tracing::{debug, trace, warn};
 
 use crate::{
     h2::{
-        parse::{Frame, FrameType, PrioritySpec},
-        types::{ConnectionClosed, H2ConnectionError},
+        parse::{parse_reserved_and_u31, Frame, FrameType, PrioritySpec},
+        types::H2ConnectionError,
     },
     util::read_and_parse,
     ExpectResponseHeaders, Headers, Method, Request, Responder, ServerDriver,
@@ -27,10 +33,7 @@ use super::{
     parse::{
         ContinuationFlags, DataFlags, HeadersFlags, PingFlags, Settings, SettingsFlags, StreamId,
     },
-    types::{
-        ConnState, ContinuationState, H2ConnEvent, H2Error, H2Result, H2StreamError, HeadersData,
-        HeadersOrTrailers, StreamStage,
-    },
+    types::{ConnState, H2ConnEvent, H2StreamError, HeadersOrTrailers, StreamState},
 };
 
 /// Reads and processes h2 frames from the client.
@@ -39,11 +42,6 @@ pub(crate) struct H2ReadContext<D: ServerDriver + 'static> {
     ev_tx: mpsc::Sender<H2ConnEvent>,
     state: ConnState,
     hpack_dec: fluke_hpack::Decoder<'static>,
-    // TODO: kill, cf. https://github.com/hapsoc/fluke/issues/121
-    continuation_state: ContinuationState,
-
-    /// Whether we've sent a GOAWAY frame.
-    pub goaway_sent: bool,
 
     /// Whether we've received a GOAWAY frame.
     pub goaway_recv: bool,
@@ -60,494 +58,476 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
             ev_tx,
             state,
             hpack_dec,
-            continuation_state: ContinuationState::Idle,
-            goaway_sent: false,
             goaway_recv: false,
         }
     }
 
     pub(crate) async fn read_loop(
-        &mut self,
-        mut client_buf: RollMut,
-        mut transport_r: impl ReadOwned,
+        mut self,
+        client_buf: RollMut,
+        transport_r: impl ReadOwned,
     ) -> eyre::Result<()> {
-        loop {
-            let frame;
-            let frame_res = read_and_parse(Frame::parse, &mut transport_r, client_buf, 128).await;
-            let maybe_frame = match frame_res {
-                Ok(inner) => inner,
-                Err(e) => {
-                    // if this is a connection reset and we've sent a goaway, ignore it
-                    if self.goaway_sent {
+        let mut goaway_err: Option<H2ConnectionError> = None;
+
+        {
+            // read frames and send them into an mpsc buffer of size 1
+            let (tx, rx) = mpsc::channel::<(Frame, Roll)>(1);
+
+            // store max frame size setting as an atomic so we can share it across tasks
+            // FIXME: the process_task should update this
+            let max_frame_size = Rc::new(AtomicU32::new(self.state.self_settings.max_frame_size));
+
+            let mut io_task =
+                std::pin::pin!(Self::io_loop(client_buf, transport_r, tx, max_frame_size));
+            let mut process_task = std::pin::pin!(self.process_loop(rx));
+
+            tokio::select! {
+                res = &mut io_task => {
+                    debug!(?res, "h2 io task finished");
+
+                    if let Err(H2ConnectionError::ReadError(e)) = res {
+                        let mut should_ignore_err = false;
+
+                        // if this is a connection reset and we've sent a goaway, ignore it
                         if let Some(io_error) = e.root_cause().downcast_ref::<std::io::Error>() {
                             if io_error.kind() == std::io::ErrorKind::ConnectionReset {
-                                debug!("ignoring connection reset after goaway");
-                                return Ok(());
+                                should_ignore_err = true;
                             }
                         }
+
+                        if !should_ignore_err {
+                            return Err(e.wrap_err("h2 io"));
+                        }
+                    }
+
+                    if let Err(e) = (&mut process_task).await {
+                        debug!("h2 process task finished with error: {e}");
+                        return Err(e).wrap_err("h2 process");
                     }
-                    return Err(e.wrap_err("h2 read"));
                 }
+                res = &mut process_task => {
+                    debug!(?res, "h2 process task finished");
+
+                    if let Err(err) = res {
+                        goaway_err = Some(err);
+                    }
+
+                    // probably don't need to wait for the io task there
+                }
+            }
+        }
+
+        if let Some(err) = goaway_err {
+            if self
+                .ev_tx
+                .send(H2ConnEvent::GoAway {
+                    err,
+                    last_stream_id: self.state.last_stream_id,
+                })
+                .await
+                .is_err()
+            {
+                debug!("error sending goaway");
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn io_loop(
+        mut client_buf: RollMut,
+        mut transport_r: impl ReadOwned,
+        tx: mpsc::Sender<(Frame, Roll)>,
+        max_frame_size: Rc<AtomicU32>,
+    ) -> Result<(), H2ConnectionError> {
+        'read_frames: loop {
+            const MAX_FRAME_HEADER_SIZE: usize = 128;
+            let frame;
+            let frame_res = read_and_parse(
+                Frame::parse,
+                &mut transport_r,
+                client_buf,
+                MAX_FRAME_HEADER_SIZE,
+            )
+            .await;
+
+            let maybe_frame = match frame_res {
+                Ok(inner) => inner,
+                Err(e) => return Err(H2ConnectionError::ReadError(e)),
             };
             (client_buf, frame) = match maybe_frame {
                 Some((client_buf, frame)) => (client_buf, frame),
                 None => {
-                    if self.goaway_sent {
-                        debug!("peer connection reset after we sent a GOAWAY");
-                        return Ok(());
-                    }
-                    if self.goaway_recv {
-                        debug!("peer connection reset after we received a GOAWAY");
-                        return Ok(());
-                    }
-
-                    debug!("peer connection reset, without a GOAWAY");
-                    return Err(eyre::Report::from(ConnectionClosed));
+                    debug!("Peer went away before sending a frame");
+                    break 'read_frames;
                 }
             };
 
             debug!(?frame, "Received");
 
-            match &frame.frame_type {
-                FrameType::Headers(..) | FrameType::Data(..) => {
-                    let max_frame_size = self.state.self_settings.max_frame_size;
-                    debug!(
-                        "frame len = {}, max_frame_size = {max_frame_size}",
-                        frame.len
-                    );
-                    if frame.len > max_frame_size {
-                        self.send_goaway(H2ConnectionError::FrameTooLarge {
-                            frame_type: frame.frame_type,
-                            frame_size: frame.len,
-                            max_frame_size,
-                        })
-                        .await;
-                        return Err(eyre::Report::from(ConnectionClosed));
-                    }
-                }
-                _ => {
-                    // muffin.
-                }
+            let max_frame_size = max_frame_size.load(Ordering::Relaxed);
+            if frame.len > max_frame_size {
+                return Err(H2ConnectionError::FrameTooLarge {
+                    frame_type: frame.frame_type,
+                    frame_size: frame.len,
+                    max_frame_size,
+                });
             }
 
-            // TODO: there might be optimizations to be done for `Data` frames later
-            // on, but for now, let's unconditionally read the payload (if it's not
-            // empty).
-            let payload: Roll = if frame.len == 0 {
-                Roll::empty()
-            } else {
-                let payload_roll;
-                (client_buf, payload_roll) = match read_and_parse(
-                    nom::bytes::streaming::take(frame.len as usize),
-                    &mut transport_r,
-                    client_buf,
-                    frame.len as usize,
-                )
-                .await?
-                {
-                    Some((client_buf, payload)) => (client_buf, payload),
-                    None => {
-                        debug!(
-                            "h2 client closed connection while reading payload for {:?}",
-                            frame.frame_type
-                        );
-                        return Ok(());
-                    }
-                };
-                payload_roll
+            let mut payload;
+            (client_buf, payload) = match read_and_parse(
+                nom::bytes::streaming::take(frame.len as usize),
+                &mut transport_r,
+                client_buf,
+                frame.len as usize,
+            )
+            .await?
+            {
+                Some((client_buf, payload)) => (client_buf, payload),
+                None => {
+                    return Err(H2ConnectionError::IncompleteFrame {
+                        frame_type: frame.frame_type,
+                        frame_size: frame.len,
+                    })
+                }
             };
 
-            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;
-                    }
+            let has_padding = match frame.frame_type {
+                FrameType::Data(flags) => flags.contains(DataFlags::Padded),
+                FrameType::Headers(flags) => flags.contains(HeadersFlags::Padded),
+                _ => false,
+            };
+
+            if has_padding {
+                if payload.is_empty() {
+                    return Err(H2ConnectionError::PaddedFrameEmpty {
+                        frame_type: frame.frame_type,
+                    });
                 }
+
+                let padding_length_roll;
+                (padding_length_roll, payload) = payload.split_at(1);
+                let padding_length = padding_length_roll[0] as usize;
+                if payload.len() < padding_length {
+                    return Err(H2ConnectionError::PaddedFrameTooShort {
+                        frame_type: frame.frame_type,
+                        padding_length,
+                        frame_size: frame.len,
+                    });
+                }
+
+                // padding is on the end of the payload
+                let at = payload.len() - padding_length;
+                (payload, _) = payload.split_at(at);
             }
-        }
-    }
 
-    // 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) -> H2Result {
-        match self.continuation_state {
-            ContinuationState::Idle => self.process_idle_frame(frame, payload).await,
-            ContinuationState::ContinuingHeaders(expected_stream_id) => {
-                // TODO: kill, cf. https://github.com/hapsoc/fluke/issues/121
-                self.process_continuation_frame(frame, payload, expected_stream_id)
-                    .await
+            if tx.send((frame, payload)).await.is_err() {
+                debug!("h2 io loop: receiver dropped, closing connection");
+                return Ok(());
             }
         }
-    }
-
-    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) {
-                    if payload.is_empty() {
-                        todo!("handle connection error: padded data frame, but no padding length");
-                    }
 
-                    let padding_length_roll;
-                    (padding_length_roll, payload) = payload.split_at(1);
-                    let padding_length = padding_length_roll[0] as usize;
-                    if payload.len() < padding_length {
-                        todo!(
-                            "handle connection error: padded headers frame, but not enough padding"
-                        );
-                    }
+        Ok(())
+    }
 
-                    let at = payload.len() - padding_length;
-                    (payload, _) = payload.split_at(at);
+    async fn process_loop(
+        &mut self,
+        mut rx: mpsc::Receiver<(Frame, Roll)>,
+    ) -> Result<(), H2ConnectionError> {
+        'process_frames: loop {
+            let (frame, mut payload) = match rx.recv().await {
+                Some(t) => t,
+                None => {
+                    debug!("h2 process task: peer hung up");
+                    break;
                 }
+            };
+
+            match frame.frame_type {
+                FrameType::Data(flags) => {
+                    let ss = self.state.streams.get_mut(&frame.stream_id).ok_or(
+                        H2ConnectionError::StreamClosed {
+                            stream_id: frame.stream_id,
+                        },
+                    )?;
+
+                    match ss {
+                        StreamState::Open(tx) => {
+                            if tx
+                                .send(Ok(PieceOrTrailers::Piece(payload.into())))
+                                .await
+                                .is_err()
+                            {
+                                warn!(
+                                    "TODO: The body is being ignored, we should reset the stream"
+                                );
+                            }
 
-                let body_tx = {
-                    let stage = self
-                        .state
-                        .streams
-                        .get_mut(&frame.stream_id)
-                        // TODO: proper error handling (connection error)
-                        .expect("received data for unknown stream");
-                    match stage {
-                        StreamStage::Headers(_) => {
-                            // TODO: proper error handling (stream error)
-                            panic!("expected headers, received data")
-                        }
-                        StreamStage::Trailers(..) => {
-                            // TODO: proper error handling (stream error)
-                            panic!("expected trailers, received data")
-                        }
-                        StreamStage::Open(tx) => {
-                            // TODO: we can get rid of that clone sometimes
-                            let tx = tx.clone();
                             if flags.contains(DataFlags::EndStream) {
-                                *stage = StreamStage::HalfClosedRemote;
+                                *ss = StreamState::HalfClosedRemote;
                             }
-                            tx
                         }
-                        StreamStage::HalfClosedRemote | StreamStage::Closed => {
-                            return H2StreamError::ReceivedDataForClosedStream
-                                .for_stream(frame.stream_id)
-                                .into();
+                        StreamState::HalfClosedRemote => {
+                            debug!(
+                                stream_id = %frame.stream_id,
+                                "Received data for closed stream"
+                            );
+                            self.send_rst(frame.stream_id, H2StreamError::StreamClosed)
+                                .await
                         }
                     }
-                };
-
-                if body_tx
-                    .send(Ok(PieceOrTrailers::Piece(payload.into())))
-                    .await
-                    .is_err()
-                {
-                    warn!("TODO: The body is being ignored, we should reset the stream");
-                }
-            }
-            FrameType::Headers(flags) => {
-                // TODO: if we're shutting down, ignore streams higher
-                // than the last one we accepted.
-
-                if frame.stream_id.is_server_initiated() {
-                    return H2ConnectionError::ClientSidShouldBeOdd.into();
                 }
+                FrameType::Headers(flags) => {
+                    if flags.contains(HeadersFlags::Priority) {
+                        let pri_spec;
+                        (payload, pri_spec) = PrioritySpec::parse(payload)
+                            .finish()
+                            .map_err(|err| eyre::eyre!("parsing error: {err:?}"))?;
+                        debug!(exclusive = %pri_spec.exclusive, stream_dependency = ?pri_spec.stream_dependency, weight = %pri_spec.weight, "received priority, exclusive");
+
+                        if pri_spec.stream_dependency == frame.stream_id {
+                            return Err(H2ConnectionError::HeadersInvalidPriority {
+                                stream_id: frame.stream_id,
+                            });
+                        }
+                    }
 
-                if frame.stream_id < self.state.last_stream_id {
-                    return H2ConnectionError::ClientSidShouldBeIncreasing.into();
-                }
-                self.state.last_stream_id = frame.stream_id;
+                    let headers_or_trailers;
+                    let mode;
+
+                    match self.state.streams.get_mut(&frame.stream_id) {
+                        None => {
+                            headers_or_trailers = HeadersOrTrailers::Headers;
+                            debug!(
+                                stream_id = %frame.stream_id,
+                                last_stream_id = %self.state.last_stream_id,
+                                next_stream_count = %self.state.streams.len() + 1,
+                                "Receiving headers",
+                            );
+
+                            if frame.stream_id.is_server_initiated() {
+                                return Err(H2ConnectionError::ClientSidShouldBeOdd);
+                            }
 
-                let padding_length = if flags.contains(HeadersFlags::Padded) {
-                    if payload.is_empty() {
-                        return H2ConnectionError::PaddedFrameEmpty.into();
+                            if frame.stream_id <= self.state.last_stream_id {
+                                debug!(
+                                    frame_stream_id = %frame.stream_id,
+                                    last_stream_id = %self.state.last_stream_id,
+                                    "Received headers for invalid stream ID"
+                                );
+
+                                // this stream may have existed, but it no longer does:
+                                return Err(H2ConnectionError::StreamClosed {
+                                    stream_id: frame.stream_id,
+                                });
+                            } else {
+                                // TODO: if we're shutting down, ignore streams higher
+                                // than the last one we accepted.
+
+                                let max_concurrent_streams =
+                                    self.state.self_settings.max_concurrent_streams;
+                                let num_streams_if_accept = self.state.streams.len() + 1;
+                                if num_streams_if_accept > max_concurrent_streams as _ {
+                                    // reset the stream, indicating we refused it
+                                    self.send_rst(frame.stream_id, H2StreamError::RefusedStream)
+                                        .await;
+
+                                    // but we still need to skip over any continuation frames
+                                    mode = ReadHeadersMode::Skip;
+                                } else {
+                                    self.state.last_stream_id = frame.stream_id;
+                                    mode = ReadHeadersMode::Process;
+                                }
+                            }
+                        }
+                        Some(StreamState::Open(_)) => {
+                            headers_or_trailers = HeadersOrTrailers::Trailers;
+                            debug!("Receiving trailers for stream {}", frame.stream_id);
+
+                            if flags.contains(HeadersFlags::EndStream) {
+                                // good, that's what we expect
+                                mode = ReadHeadersMode::Process;
+                            } else {
+                                // ignore trailers, we're not accepting the stream
+                                mode = ReadHeadersMode::Skip;
+
+                                self.send_rst(frame.stream_id, H2StreamError::TrailersNotEndStream)
+                                    .await
+                            }
+                        }
+                        Some(StreamState::HalfClosedRemote) => {
+                            return Err(H2ConnectionError::StreamClosed {
+                                stream_id: frame.stream_id,
+                            });
+                        }
                     }
 
-                    let padding_length_roll;
-                    (padding_length_roll, payload) = payload.split_at(1);
-                    padding_length_roll[0] as usize
-                } else {
-                    0
-                };
-
-                if flags.contains(HeadersFlags::Priority) {
-                    let pri_spec;
-                    (payload, pri_spec) = PrioritySpec::parse(payload)
-                        .finish()
-                        .map_err(|err| eyre::eyre!("parsing error: {err:?}"))?;
-                    debug!(exclusive = %pri_spec.exclusive, stream_dependency = ?pri_spec.stream_dependency, weight = %pri_spec.weight, "received priority, exclusive");
+                    self.read_headers(
+                        headers_or_trailers,
+                        mode,
+                        flags,
+                        frame.stream_id,
+                        payload,
+                        &mut rx,
+                    )
+                    .await?;
+                }
+                FrameType::Priority => {
+                    let pri_spec = match PrioritySpec::parse(payload) {
+                        Ok((_rest, pri_spec)) => pri_spec,
+                        Err(e) => {
+                            todo!("handle connection error: invalid priority frame {e}")
+                        }
+                    };
+                    debug!(?pri_spec, "received priority frame");
 
                     if pri_spec.stream_dependency == frame.stream_id {
-                        self.send_goaway(H2ConnectionError::HeadersInvalidPriority {
+                        return Err(H2ConnectionError::HeadersInvalidPriority {
                             stream_id: frame.stream_id,
-                        })
-                        .await;
-                        return Ok(());
+                        });
                     }
                 }
-
-                if padding_length > 0 {
-                    if payload.len() < padding_length {
-                        self.send_goaway(H2ConnectionError::PaddedFrameTooShort)
-                            .await;
+                FrameType::RstStream => match self.state.streams.remove(&frame.stream_id) {
+                    None => {
+                        return Err(H2ConnectionError::RstStreamForUnknownStream {
+                            stream_id: frame.stream_id,
+                        })
                     }
-
-                    let at = payload.len() - padding_length;
-                    (payload, _) = payload.split_at(at);
-                }
-
-                let headers_data = HeadersData {
-                    end_stream: flags.contains(HeadersFlags::EndStream),
-                    fragments: smallvec![payload],
-                };
-
-                match self.state.streams.get_mut(&frame.stream_id) {
-                    Some(stage) => {
-                        debug!("Receiving trailers for stream {}", frame.stream_id);
-
-                        if !flags.contains(HeadersFlags::EndStream) {
-                            todo!(
-                                            "handle connection error: trailers must have EndStream, this just looks like duplicate headers for stream {}",
-                                            frame.stream_id
-                                        );
+                    Some(ss) => match ss {
+                        StreamState::Open(body_tx) => {
+                            _ = body_tx
+                                .send(Err(H2StreamError::ReceivedRstStream.into()))
+                                .await;
                         }
-
-                        let prev_stage = std::mem::replace(stage, StreamStage::HalfClosedRemote);
-                        match prev_stage {
-                            StreamStage::Open(body_tx) => {
-                                *stage = StreamStage::Trailers(body_tx, headers_data);
-                            }
-                            // FIXME: that's a connection error
-                            _ => unreachable!(),
+                        StreamState::HalfClosedRemote => {
+                            // good
                         }
+                    },
+                },
+                FrameType::Settings(s) => {
+                    if frame.stream_id != StreamId::CONNECTION {
+                        return Err(H2ConnectionError::SettingsWithNonZeroStreamId {
+                            stream_id: frame.stream_id,
+                        });
                     }
-                    None => {
-                        debug!(
-                            "Receiving headers for stream {} (if we accept this stream we'll have {} total)",
-                            frame.stream_id,
-                            self.state.streams.len() + 1
-                        );
-
-                        let num_streams_if_accept = self.state.streams.len() + 1;
-                        if num_streams_if_accept
-                            > self.state.self_settings.max_concurrent_streams as _
-                        {
-                            self.send_goaway(H2ConnectionError::MaxConcurrentStreamsExceeded {
-                                max_concurrent_streams: self
-                                    .state
-                                    .self_settings
-                                    .max_concurrent_streams,
+
+                    if s.contains(SettingsFlags::Ack) {
+                        debug!("Peer has acknowledged our settings, cool");
+                        if !payload.is_empty() {
+                            return Err(H2ConnectionError::SettingsAckWithPayload {
+                                len: payload.len() as _,
+                            });
+                        }
+                    } else {
+                        let (_, settings) = Settings::parse(payload)
+                            .finish()
+                            .map_err(|err| eyre::eyre!("parsing error: {err:?}"))?;
+                        let new_max_header_table_size = settings.header_table_size;
+                        debug!(?settings, "Received settings");
+                        self.state.peer_settings = settings;
+
+                        if self
+                            .ev_tx
+                            .send(H2ConnEvent::AcknowledgeSettings {
+                                new_max_header_table_size,
                             })
-                            .await;
-                            return Ok(());
+                            .await
+                            .is_err()
+                        {
+                            return Err(eyre::eyre!(
+                                "could not send H2 acknowledge settings event"
+                            )
+                            .into());
                         }
-                        self.state
-                            .streams
-                            .insert(frame.stream_id, StreamStage::Headers(headers_data));
                     }
                 }
-
-                if flags.contains(HeadersFlags::EndHeaders) {
-                    self.end_headers(frame.stream_id).await?;
-                } else {
-                    debug!(
-                        "expecting more headers/trailers for stream {}",
-                        frame.stream_id
-                    );
-                    self.continuation_state = ContinuationState::ContinuingHeaders(frame.stream_id);
+                FrameType::PushPromise => {
+                    return Err(H2ConnectionError::ClientSentPushPromise);
                 }
-            }
-            FrameType::Priority => {
-                let pri_spec = match PrioritySpec::parse(payload) {
-                    Ok((_rest, pri_spec)) => pri_spec,
-                    Err(e) => {
-                        todo!("handle connection error: invalid priority frame {e}")
+                FrameType::Ping(flags) => {
+                    if frame.stream_id != StreamId::CONNECTION {
+                        return Err(H2ConnectionError::PingFrameWithNonZeroStreamId {
+                            stream_id: frame.stream_id,
+                        });
                     }
-                };
-                debug!(?pri_spec, "received priority frame");
 
-                if pri_spec.stream_dependency == frame.stream_id {
-                    self.send_goaway(H2ConnectionError::HeadersInvalidPriority {
-                        stream_id: frame.stream_id,
-                    })
-                    .await;
-                    return Ok(());
-                }
-            }
-            FrameType::RstStream => todo!("implement RstStream"),
-            FrameType::Settings(s) => {
-                if s.contains(SettingsFlags::Ack) {
-                    debug!("Peer has acknowledged our settings, cool");
-                } else {
-                    // TODO: actually apply settings
-                    let (_, settings) = Settings::parse(payload)
-                        .finish()
-                        .map_err(|err| eyre::eyre!("parsing error: {err:?}"))?;
-                    let new_max_header_table_size = settings.header_table_size;
-                    debug!(?settings, "Received settings");
-                    self.state.peer_settings = settings;
-
-                    if self
-                        .ev_tx
-                        .send(H2ConnEvent::AcknowledgeSettings {
-                            new_max_header_table_size,
-                        })
-                        .await
-                        .is_err()
-                    {
-                        return Err(
-                            eyre::eyre!("could not send H2 acknowledge settings event").into()
-                        );
+                    if frame.len != 8 {
+                        return Err(H2ConnectionError::PingFrameInvalidLength { len: frame.len });
                     }
-                }
-            }
-            FrameType::PushPromise => {
-                self.send_goaway(H2ConnectionError::ClientSentPushPromise)
-                    .await;
-                return Ok(());
-            }
-            FrameType::Ping(flags) => {
-                if frame.stream_id != StreamId::CONNECTION {
-                    todo!("handle connection error: ping frame with non-zero stream id");
-                }
-
-                if frame.len != 8 {
-                    todo!("handle connection error: ping frame with invalid length");
-                }
 
-                if flags.contains(PingFlags::Ack) {
-                    // TODO: check that payload matches the one we sent?
-
-                    debug!("received ping ack");
-                    return Ok(());
-                }
-
-                if self.ev_tx.send(H2ConnEvent::Ping(payload)).await.is_err() {
-                    return Err(eyre::eyre!("could not send H2 ping event").into());
-                }
-            }
-            FrameType::GoAway => {
-                self.goaway_recv = true;
+                    if flags.contains(PingFlags::Ack) {
+                        // TODO: check that payload matches the one we sent?
+                        continue 'process_frames;
+                    }
 
-                // TODO: this should probably have other effects than setting
-                // this flag.
-            }
-            FrameType::WindowUpdate => match frame.stream_id.0 {
-                0 => {
-                    debug!("TODO: ignoring connection-wide window update");
-                }
-                _ => match self.state.streams.get_mut(&frame.stream_id) {
-                    Some(_ss) => {
-                        debug!("TODO: handle window update for stream {}", frame.stream_id)
+                    if self.ev_tx.send(H2ConnEvent::Ping(payload)).await.is_err() {
+                        return Err(eyre::eyre!("could not send H2 ping event").into());
                     }
-                    None => {
-                        self.send_goaway(H2ConnectionError::WindowUpdateForUnknownStream {
+                }
+                FrameType::GoAway => {
+                    if frame.stream_id != StreamId::CONNECTION {
+                        return Err(H2ConnectionError::GoAwayWithNonZeroStreamId {
                             stream_id: frame.stream_id,
-                        })
-                        .await;
+                        });
                     }
-                },
-            },
-            FrameType::Continuation(_flags) => {
-                self.send_goaway(H2ConnectionError::UnexpectedContinuationFrame {
-                    stream_id: frame.stream_id,
-                })
-                .await;
-            }
-            FrameType::Unknown(ft) => {
-                trace!(
-                    "ignoring unknown frame with type 0x{:x}, flags 0x{:x}",
-                    ft.ty,
-                    ft.flags
-                );
-            }
-        }
 
-        Ok(())
-    }
+                    self.goaway_recv = true;
 
-    async fn process_continuation_frame(
-        &mut self,
-        frame: Frame,
-        payload: Roll,
-        expected_stream_id: StreamId,
-    ) -> H2Result {
-        let flags = match frame.frame_type {
-            FrameType::Continuation(flags) => flags,
-            other => {
-                return H2ConnectionError::ExpectedContinuationFrame {
-                    stream_id: expected_stream_id,
-                    frame_type: other,
+                    // TODO: this should probably have other effects than setting
+                    // this flag.
                 }
-                .into()
-            }
-        };
-
-        if frame.stream_id != expected_stream_id {
-            return H2ConnectionError::ExpectedContinuationForStream {
-                stream_id: expected_stream_id,
-                continuation_stream_id: frame.stream_id,
-            }
-            .into();
-        }
-
-        // unwrap rationale: we just checked that this is a
-        // continuation of a stream we've already learned about.
-        let ss = self.state.streams.get_mut(&frame.stream_id).unwrap();
-        match ss {
-            StreamStage::Headers(data) | StreamStage::Trailers(_, data) => {
-                data.fragments.push(payload);
-            }
-            _ => {
-                // FIXME: store `HeadersData` in
-                // `ContinuationState` directly so this
-                // branch doesn't even exist.
-                unreachable!()
-            }
-        }
+                FrameType::WindowUpdate => {
+                    if payload.len() != 4 {
+                        return Err(H2ConnectionError::WindowUpdateInvalidLength {
+                            len: payload.len() as _,
+                        });
+                    }
 
-        if flags.contains(ContinuationFlags::EndHeaders) {
-            self.end_headers(frame.stream_id).await
-        } else {
-            debug!(
-                "expecting more headers/trailers for stream {}",
-                frame.stream_id
-            );
-            Ok(())
-        }
-    }
+                    let increment;
+                    (_, (_, increment)) = parse_reserved_and_u31(payload)
+                        .finish()
+                        .map_err(|err| eyre::eyre!("parsing error: {err:?}"))?;
 
-    async fn send_goaway(&mut self, err: H2ConnectionError) {
-        // TODO: this should change the global server state: we should ignore
-        // any streams higher than the last stream id we've seen after
-        // we've done that.
+                    if increment == 0 {
+                        return Err(H2ConnectionError::WindowUpdateZeroIncrement);
+                    }
 
-        if self.goaway_sent {
-            debug!("goaway already sent, ignoring");
-            return;
+                    if frame.stream_id == StreamId::CONNECTION {
+                        debug!("TODO: ignoring connection-wide window update");
+                    } else {
+                        match self.state.streams.get_mut(&frame.stream_id) {
+                            None => {
+                                return Err(H2ConnectionError::WindowUpdateForUnknownStream {
+                                    stream_id: frame.stream_id,
+                                });
+                            }
+                            Some(_ss) => {
+                                debug!("TODO: handle window update for stream {}", frame.stream_id)
+                            }
+                        }
+                    }
+                }
+                FrameType::Continuation(_flags) => {
+                    return Err(H2ConnectionError::UnexpectedContinuationFrame {
+                        stream_id: frame.stream_id,
+                    });
+                }
+                FrameType::Unknown(ft) => {
+                    trace!(
+                        "ignoring unknown frame with type 0x{:x}, flags 0x{:x}",
+                        ft.ty,
+                        ft.flags
+                    );
+                }
+            }
         }
 
-        self.goaway_sent = true;
-        if self
-            .ev_tx
-            .send(H2ConnEvent::GoAway {
-                err,
-                last_stream_id: self.state.last_stream_id,
-            })
-            .await
-            .is_err()
-        {
-            debug!("error sending goaway");
-        }
+        Ok(())
     }
 
     async fn send_rst(&mut self, stream_id: StreamId, e: H2StreamError) {
         let error_code = e.as_known_error_code();
-        warn!("sending rst because: {e} (known error code: {error_code:?})");
+        debug!("Sending rst because: {e} (known error code: {error_code:?})");
 
         if self
             .ev_tx
@@ -564,17 +544,81 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
         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(&mut self, stream_id: StreamId) -> H2Result {
-        let stage = self
-            .state
-            .streams
-            .get_mut(&stream_id)
-            // FIXME: don't panic
-            .expect("stream stage must exist");
+    async fn read_headers(
+        &mut self,
+        headers_or_trailers: HeadersOrTrailers,
+        mode: ReadHeadersMode,
+        flags: BitFlags<HeadersFlags, u8>,
+        stream_id: StreamId,
+        payload: Roll,
+        rx: &mut mpsc::Receiver<(Frame, Roll)>,
+    ) -> Result<(), H2ConnectionError> {
+        let end_stream = flags.contains(HeadersFlags::EndStream);
+
+        enum Data {
+            Single(Roll),
+            Multi(SmallVec<[Roll; 2]>),
+        }
+
+        let data = if flags.contains(HeadersFlags::EndHeaders) {
+            // good, no continuation frames needed
+            Data::Single(payload)
+        } else {
+            // read continuation frames
+
+            #[allow(unused, clippy::let_unit_value)]
+            let flags = (); // don't accidentally use the `flags` variable
+
+            let mut fragments = smallvec![payload];
+
+            loop {
+                let (continuation_frame, continuation_payload) = match rx.recv().await {
+                    Some(t) => t,
+                    None => {
+                        // even though this error is "for a stream", it's a
+                        // connection error, because it means the peer doesn't
+                        // know how to speak HTTP/2.
+                        return Err(H2ConnectionError::ExpectedContinuationFrame {
+                            stream_id,
+                            frame_type: None,
+                        });
+                    }
+                };
+
+                if stream_id != continuation_frame.stream_id {
+                    return Err(H2ConnectionError::ExpectedContinuationForStream {
+                        stream_id,
+                        continuation_stream_id: continuation_frame.stream_id,
+                    });
+                }
+
+                let cont_flags = match continuation_frame.frame_type {
+                    FrameType::Continuation(flags) => flags,
+                    other => {
+                        return Err(H2ConnectionError::ExpectedContinuationFrame {
+                            stream_id,
+                            frame_type: Some(other),
+                        })
+                    }
+                };
+
+                // add fragment
+                fragments.push(continuation_payload);
+
+                if cont_flags.contains(ContinuationFlags::EndHeaders) {
+                    // we're done
+                    break;
+                }
+            }
+
+            Data::Multi(fragments)
+        };
+
+        if matches!(mode, ReadHeadersMode::Skip) {
+            // that's all we need to do: we're not actually validating the
+            // headers, we already send a RST
+            return Ok(());
+        }
 
         let mut method: Option<Method> = None;
         let mut scheme: Option<Scheme> = None;
@@ -583,102 +627,90 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
 
         let mut headers = Headers::default();
 
-        let mut decode_data =
-            |headers_or_trailers: HeadersOrTrailers, data: &HeadersData| -> H2Result {
-                // TODO: find a way to propagate errors from here - probably will have to change
-                // the function signature in fluke-hpack.
-                let on_header_pair = |key: Cow<[u8]>, value: Cow<[u8]>| {
-                    debug!(
-                        "{headers_or_trailers:?} | {}: {}",
-                        std::str::from_utf8(&key).unwrap_or("<non-utf8-key>"), // TODO: does this hurt performance when debug logging is disabled?
-                        std::str::from_utf8(&value).unwrap_or("<non-utf8-value>"),
-                    );
+        // TODO: find a way to propagate errors from here - probably will have to change
+        // the function signature in fluke-hpack, or just write to some captured
+        // error
+        let on_header_pair = |key: Cow<[u8]>, value: Cow<[u8]>| {
+            debug!(
+                "{headers_or_trailers:?} | {}: {}",
+                std::str::from_utf8(&key).unwrap_or("<non-utf8-key>"), // TODO: does this hurt performance when debug logging is disabled?
+                std::str::from_utf8(&value).unwrap_or("<non-utf8-value>"),
+            );
 
-                    if &key[..1] == b":" {
-                        if matches!(headers_or_trailers, HeadersOrTrailers::Trailers) {
-                            // TODO: proper error handling
-                            panic!("trailers cannot contain pseudo-headers");
-                        }
+            if &key[..1] == b":" {
+                if matches!(headers_or_trailers, HeadersOrTrailers::Trailers) {
+                    // TODO: proper error handling
+                    panic!("trailers cannot contain pseudo-headers");
+                }
 
-                        // it's a pseudo-header!
-                        // TODO: reject headers that occur after pseudo-headers
-                        match &key[1..] {
-                            b"method" => {
-                                // TODO: error handling
-                                let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
-                                if method.replace(Method::from(value)).is_some() {
-                                    unreachable!(); // No duplicate allowed.
-                                }
-                            }
-                            b"scheme" => {
-                                // TODO: error handling
-                                let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
-                                if scheme.replace(value.parse().unwrap()).is_some() {
-                                    unreachable!(); // No duplicate allowed.
-                                }
-                            }
-                            b"path" => {
-                                // TODO: error handling
-                                let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
-                                if value.len() == 0 || path.replace(value).is_some() {
-                                    unreachable!(); // No empty path nor duplicate allowed.
-                                }
-                            }
-                            b"authority" => {
-                                // TODO: error handling
-                                let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
-                                if authority.replace(value.parse().unwrap()).is_some() {
-                                    unreachable!(); // No duplicate allowed. (h2spec doesn't seem to test for
-                                                    // this case but rejecting for duplicates seems
-                                                    // reasonable.)
-                                }
-                            }
-                            _ => {
-                                debug!("ignoring pseudo-header");
-                            }
+                // it's a pseudo-header!
+                // TODO: reject headers that occur after pseudo-headers
+                match &key[1..] {
+                    b"method" => {
+                        // TODO: error handling
+                        let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
+                        if method.replace(Method::from(value)).is_some() {
+                            unreachable!(); // No duplicate allowed.
                         }
-                    } else {
-                        // TODO: what do we do in case of malformed header names?
-                        // ignore it? return a 400?
-                        let name = HeaderName::from_bytes(&key[..]).expect("malformed header name");
-                        let value: Piece = value.to_vec().into();
-                        headers.append(name, value);
                     }
-                };
-
-                match &data.fragments[..] {
-                    [] => unreachable!("must have at least one fragment"),
-                    [payload] => {
-                        self.hpack_dec
-                            .decode_with_cb(&payload[..], on_header_pair)
-                            .map_err(|e| H2ConnectionError::CompressionError(format!("{e:?}")))?;
+                    b"scheme" => {
+                        // TODO: error handling
+                        let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
+                        if scheme.replace(value.parse().unwrap()).is_some() {
+                            unreachable!(); // No duplicate allowed.
+                        }
                     }
-                    _ => {
-                        let total_len = data.fragments.iter().map(|f| f.len()).sum();
-                        // this is a slow path, let's do a little heap allocation. we could
-                        // be using `RollMut` for this, but it would probably need to resize
-                        // a bunch
-                        let mut payload = Vec::with_capacity(total_len);
-                        for frag in &data.fragments {
-                            payload.extend_from_slice(&frag[..]);
+                    b"path" => {
+                        // TODO: error handling
+                        let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
+                        if value.len() == 0 || path.replace(value).is_some() {
+                            unreachable!(); // No empty path nor duplicate allowed.
                         }
-                        self.hpack_dec
-                            .decode_with_cb(&payload[..], on_header_pair)
-                            .map_err(|e| H2ConnectionError::CompressionError(format!("{e:?}")))?;
                     }
-                };
-
-                Ok(())
-            };
-
-        // FIXME: don't panic on unexpected states here
-        let mut prev_stage = StreamStage::Closed;
-        std::mem::swap(&mut prev_stage, stage);
+                    b"authority" => {
+                        // TODO: error handling
+                        let value: PieceStr = Piece::from(value.to_vec()).to_str().unwrap();
+                        if authority.replace(value.parse().unwrap()).is_some() {
+                            unreachable!(); // No duplicate allowed. (h2spec doesn't seem to test for
+                                            // this case but rejecting duplicates seems reasonable.)
+                        }
+                    }
+                    _ => {
+                        debug!("ignoring pseudo-header");
+                    }
+                }
+            } else {
+                // TODO: what do we do in case of malformed header names?
+                // ignore it? return a 400?
+                let name = HeaderName::from_bytes(&key[..]).expect("malformed header name");
+                let value: Piece = value.to_vec().into();
+                headers.append(name, value);
+            }
+        };
 
-        match prev_stage {
-            StreamStage::Headers(data) => {
-                decode_data(HeadersOrTrailers::Headers, &data)?;
+        match data {
+            Data::Single(payload) => {
+                self.hpack_dec
+                    .decode_with_cb(&payload[..], on_header_pair)
+                    .map_err(|e| H2ConnectionError::CompressionError(format!("{e:?}")))?;
+            }
+            Data::Multi(fragments) => {
+                let total_len = fragments.iter().map(|f| f.len()).sum();
+                // this is a slow path, let's do a little heap allocation. we could
+                // be using `RollMut` for this, but it would probably need to resize
+                // a bunch
+                let mut payload = Vec::with_capacity(total_len);
+                for frag in &fragments {
+                    payload.extend_from_slice(&frag[..]);
+                }
+                self.hpack_dec
+                    .decode_with_cb(&payload[..], on_header_pair)
+                    .map_err(|e| H2ConnectionError::CompressionError(format!("{e:?}")))?;
+            }
+        };
 
+        match headers_or_trailers {
+            HeadersOrTrailers::Headers => {
                 // TODO: cf. https://httpwg.org/specs/rfc9113.html#HttpRequest
                 // A server SHOULD treat a request as malformed if it contains a Host header
                 // field that identifies an entity that differs from the entity in the
@@ -729,8 +761,8 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                 let req_body = H2Body {
                     // FIXME: that's not right. h2 requests can still specify
                     // a content-length
-                    content_length: if data.end_stream { Some(0) } else { None },
-                    eof: data.end_stream,
+                    content_length: if end_stream { Some(0) } else { None },
+                    eof: end_stream,
                     rx: piece_rx,
                 };
 
@@ -752,28 +784,43 @@ impl<D: ServerDriver + 'static> H2ReadContext<D> {
                     }
                 });
 
-                *stage = if data.end_stream {
-                    StreamStage::HalfClosedRemote
-                } else {
-                    StreamStage::Open(piece_tx)
-                };
+                self.state.streams.insert(
+                    stream_id,
+                    if end_stream {
+                        StreamState::HalfClosedRemote
+                    } else {
+                        StreamState::Open(piece_tx)
+                    },
+                );
             }
-            StreamStage::Open(_) => unreachable!(),
-            StreamStage::Trailers(body_tx, data) => {
-                decode_data(HeadersOrTrailers::Trailers, &data)?;
-
-                if body_tx
-                    .send(Ok(PieceOrTrailers::Trailers(Box::new(headers))))
-                    .await
-                    .is_err()
-                {
-                    warn!("TODO: The body is being ignored, we should reset the stream");
+            HeadersOrTrailers::Trailers => {
+                match self.state.streams.get_mut(&stream_id) {
+                    Some(StreamState::Open(body_tx)) => {
+                        if body_tx
+                            .send(Ok(PieceOrTrailers::Trailers(Box::new(headers))))
+                            .await
+                            .is_err()
+                        {
+                            // the body is being ignored, but there's no point in
+                            // resetting the stream since we just got the end of it
+                        }
+                    }
+                    _ => {
+                        unreachable!("stream state should be open when we receive trailers")
+                    }
                 }
+                self.state.streams.remove(&stream_id);
             }
-            StreamStage::HalfClosedRemote => unreachable!(),
-            StreamStage::Closed => unreachable!(),
         }
 
         Ok(())
     }
 }
+
+enum ReadHeadersMode {
+    // we're accepting the stream or processing trailers, we want to
+    // process the headers we read.
+    Process,
+    // we're refusing the stream, we want to skip over the headers we read.
+    Skip,
+}
diff --git a/crates/fluke/src/h2/server.rs b/crates/fluke/src/h2/server.rs
index 9dff2572..32a0dbc5 100644
--- a/crates/fluke/src/h2/server.rs
+++ b/crates/fluke/src/h2/server.rs
@@ -70,14 +70,7 @@ pub async fn serve(
 
     let (ev_tx, ev_rx) = tokio::sync::mpsc::channel::<H2ConnEvent>(32);
 
-    let mut h2_read_cx = H2ReadContext::new(driver.clone(), ev_tx.clone(), state);
-
-    enum Outcome {
-        PeerGone,
-        Ok,
-    }
-
-    let mut outcome = Outcome::Ok;
+    let h2_read_cx = H2ReadContext::new(driver.clone(), ev_tx, state);
 
     {
         let mut read_task = std::pin::pin!(h2_read_cx.read_loop(client_buf, transport_r));
@@ -88,7 +81,7 @@ pub async fn serve(
             res = &mut read_task => {
                 match res {
                     Err(e) => {
-                        return Err(e.wrap_err("h2 read (finished first)"));
+                       return Err(e.wrap_err("h2 read (finished first)"));
                     }
                     Ok(()) => {
                         debug!("read task finished, waiting on write task now");
@@ -96,7 +89,7 @@ pub async fn serve(
                         match res {
                             Err(e) => {
                                 if is_peer_gone(&e) {
-                                    outcome = Outcome::PeerGone;
+                                    debug!(%e, "write task failed with peer gone");
                                 } else {
                                     return Err(e.wrap_err("h2 write (finished second)"));
                                 }
@@ -112,7 +105,7 @@ pub async fn serve(
                 match res {
                     Err(e) => {
                         if is_peer_gone(&e) {
-                            outcome = Outcome::PeerGone;
+                            debug!(%e, "write task failed with peer gone");
                         } else {
                             return Err(e.wrap_err("h2 write (finished first)"));
                         }
@@ -125,21 +118,7 @@ pub async fn serve(
         };
     };
 
-    match outcome {
-        Outcome::PeerGone => {
-            if h2_read_cx.goaway_sent {
-                debug!("write task failed with broken pipe, but we already sent a goaway, so we're good");
-            } else {
-                return Err(eyre::eyre!(
-                    "peer closed connection without sending a goaway"
-                ));
-            }
-        }
-        Outcome::Ok => {
-            // all goodt hen!
-        }
-    }
-
+    debug!("finished serving");
     Ok(())
 }
 
diff --git a/crates/fluke/src/h2/types.rs b/crates/fluke/src/h2/types.rs
index a7ce2dc0..1ae01e95 100644
--- a/crates/fluke/src/h2/types.rs
+++ b/crates/fluke/src/h2/types.rs
@@ -1,7 +1,6 @@
 use std::{collections::HashMap, fmt};
 
 use fluke_buffet::{Piece, Roll};
-use smallvec::SmallVec;
 
 use crate::Response;
 
@@ -10,15 +9,8 @@ use super::{
     parse::{FrameType, KnownErrorCode, Settings, StreamId},
 };
 
-#[derive(Default, Clone, Copy)]
-pub(crate) enum ContinuationState {
-    #[default]
-    Idle,
-    ContinuingHeaders(StreamId),
-}
-
 pub(crate) struct ConnState {
-    pub(crate) streams: HashMap<StreamId, StreamStage>,
+    pub(crate) streams: HashMap<StreamId, StreamState>,
     pub(crate) last_stream_id: StreamId,
     pub(crate) self_settings: Settings,
     pub(crate) peer_settings: Settings,
@@ -75,25 +67,10 @@ impl Default for ConnState {
 //  R:  RST_STREAM frame
 //  PP:  PUSH_PROMISE frame (with implied CONTINUATION frames); state
 //     transitions are for the promised stream
-//
-// FIXME: why tf is this called "StreamStage", the RFC calls it "Stream State".
-pub(crate) enum StreamStage {
-    // TODO: kill this variant: we should be going from Idle to Open directly,
-    // cf. https://github.com/hapsoc/fluke/issues/121
-    Headers(HeadersData),
+pub(crate) enum StreamState {
     Open(H2BodySender),
-    // TODO: kill this variant too, see 121 above
-    Trailers(H2BodySender, HeadersData),
     HalfClosedRemote,
-    Closed,
-}
-
-pub(crate) struct HeadersData {
-    /// If true, no DATA frames follow, cf. https://httpwg.org/specs/rfc9113.html#HttpFraming
-    pub(crate) end_stream: bool,
-
-    /// The field block fragments
-    pub(crate) fragments: SmallVec<[Roll; 2]>,
+    // note: the "Closed" state is indicated by not having an entry in the map
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -105,25 +82,32 @@ pub(crate) enum H2ConnectionError {
         max_frame_size: u32,
     },
 
+    #[error("remote hung up while reading payload of {frame_type:?} with length {frame_size}")]
+    IncompleteFrame {
+        frame_type: FrameType,
+        frame_size: u32,
+    },
+
     #[error("headers frame had invalid priority: stream {stream_id} depends on itself")]
     HeadersInvalidPriority { stream_id: StreamId },
 
     #[error("client tried to initiate an even-numbered stream")]
     ClientSidShouldBeOdd,
 
-    #[error("client is trying to initiate stream with ID lower than the last one it initiated")]
-    ClientSidShouldBeIncreasing,
-
-    #[error("received frame with Padded flag but empty payload")]
-    PaddedFrameEmpty,
+    #[error("received {frame_type:?} frame with Padded flag but empty payload")]
+    PaddedFrameEmpty { frame_type: FrameType },
 
-    #[error("received frame with Padded flag but payload too short to contain padding length")]
-    PaddedFrameTooShort,
+    #[error("received {frame_type:?} with Padded flag but payload was shorter than padding")]
+    PaddedFrameTooShort {
+        frame_type: FrameType,
+        padding_length: usize,
+        frame_size: u32,
+    },
 
     #[error("on stream {stream_id}, expected continuation frame, but got {frame_type:?}")]
     ExpectedContinuationFrame {
         stream_id: StreamId,
-        frame_type: FrameType,
+        frame_type: Option<FrameType>,
     },
 
     #[error("expected continuation from for stream {stream_id}, but got continuation for stream {continuation_stream_id}")]
@@ -145,23 +129,38 @@ pub(crate) enum H2ConnectionError {
     #[error("received window update for unknown stream {stream_id}")]
     WindowUpdateForUnknownStream { stream_id: StreamId },
 
-    #[error("max concurrent streams exceeded (more than {max_concurrent_streams})")]
-    MaxConcurrentStreamsExceeded { max_concurrent_streams: u32 },
-
     #[error("other error: {0:?}")]
     Internal(#[from] eyre::Report),
-}
 
-impl<T> From<H2ConnectionError> for H2Result<T> {
-    fn from(e: H2ConnectionError) -> Self {
-        Err(e.into())
-    }
-}
+    #[error("error reading/parsing H2 frame: {0:?}")]
+    ReadError(eyre::Report),
 
-impl<T> From<H2Error> for H2Result<T> {
-    fn from(e: H2Error) -> Self {
-        Err(e)
-    }
+    #[error("received rst frame for unknown stream")]
+    RstStreamForUnknownStream { stream_id: StreamId },
+
+    #[error("received frame for closed stream {stream_id}")]
+    StreamClosed { stream_id: StreamId },
+
+    #[error("received ping frame frame with non-zero stream id")]
+    PingFrameWithNonZeroStreamId { stream_id: StreamId },
+
+    #[error("received ping frame with invalid length {len}")]
+    PingFrameInvalidLength { len: u32 },
+
+    #[error("received settings frame with invalid length {len}")]
+    SettingsAckWithPayload { len: u32 },
+
+    #[error("received settings frame with non-zero stream id")]
+    SettingsWithNonZeroStreamId { stream_id: StreamId },
+
+    #[error("received goaway frame with non-zero stream id")]
+    GoAwayWithNonZeroStreamId { stream_id: StreamId },
+
+    #[error("zero increment in window update frame for stream")]
+    WindowUpdateZeroIncrement,
+
+    #[error("received window update frame with invalid length {len}")]
+    WindowUpdateInvalidLength { len: usize },
 }
 
 impl H2ConnectionError {
@@ -169,10 +168,15 @@ impl H2ConnectionError {
         match self {
             // frame size errors
             H2ConnectionError::FrameTooLarge { .. } => KnownErrorCode::FrameSizeError,
-            H2ConnectionError::PaddedFrameEmpty => KnownErrorCode::FrameSizeError,
-            H2ConnectionError::PaddedFrameTooShort => KnownErrorCode::FrameSizeError,
+            H2ConnectionError::PaddedFrameEmpty { .. } => KnownErrorCode::FrameSizeError,
+            H2ConnectionError::PaddedFrameTooShort { .. } => KnownErrorCode::FrameSizeError,
+            H2ConnectionError::PingFrameInvalidLength { .. } => KnownErrorCode::FrameSizeError,
+            H2ConnectionError::SettingsAckWithPayload { .. } => KnownErrorCode::FrameSizeError,
+            H2ConnectionError::WindowUpdateInvalidLength { .. } => KnownErrorCode::FrameSizeError,
             // compression errors
             H2ConnectionError::CompressionError(_) => KnownErrorCode::CompressionError,
+            // stream closed error
+            H2ConnectionError::StreamClosed { .. } => KnownErrorCode::StreamClosed,
             // internal errors
             H2ConnectionError::Internal(_) => KnownErrorCode::InternalError,
             // protocol errors
@@ -181,23 +185,6 @@ impl H2ConnectionError {
     }
 }
 
-pub(crate) type H2Result<T = ()> = Result<T, H2Error>;
-
-#[derive(Debug, thiserror::Error)]
-pub(crate) enum H2Error {
-    #[error("connection error: {0:?}")]
-    Connection(#[from] H2ConnectionError),
-
-    #[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)]
@@ -207,8 +194,17 @@ pub(crate) enum H2StreamError {
         content_length: u64,
     },
 
-    #[error("received data for closed stream")]
-    ReceivedDataForClosedStream,
+    #[error("refused stream (would exceed max concurrent streams)")]
+    RefusedStream,
+
+    #[error("trailers must have EndStream flag set")]
+    TrailersNotEndStream,
+
+    #[error("received RST_STREAM frame")]
+    ReceivedRstStream,
+
+    #[error("stream closed")]
+    StreamClosed,
 }
 
 impl H2StreamError {
@@ -217,14 +213,11 @@ impl H2StreamError {
         use KnownErrorCode as Code;
 
         match self {
-            DataLengthDoesNotMatchContentLength { .. } => Code::ProtocolError,
-            ReceivedDataForClosedStream => Code::StreamClosed,
+            StreamClosed => Code::StreamClosed,
+            RefusedStream => Code::RefusedStream,
+            _ => Code::ProtocolError,
         }
     }
-
-    pub(crate) fn for_stream(self, stream_id: StreamId) -> H2Error {
-        H2Error::Stream(stream_id, self)
-    }
 }
 
 #[derive(Debug, Clone, Copy)]
diff --git a/crates/fluke/src/h2/write.rs b/crates/fluke/src/h2/write.rs
index a646aec1..aef86958 100644
--- a/crates/fluke/src/h2/write.rs
+++ b/crates/fluke/src/h2/write.rs
@@ -1,7 +1,7 @@
 use enumflags2::BitFlags;
 use eyre::Context;
 use tokio::sync::mpsc;
-use tracing::{debug, trace, warn};
+use tracing::{debug, trace};
 
 use crate::h2::{
     parse::{DataFlags, Frame, FrameType, HeadersFlags, PingFlags, SettingsFlags, StreamId},
@@ -115,7 +115,7 @@ pub(crate) async fn h2_write_loop(
                 last_stream_id,
             } => {
                 let error_code = err.as_known_error_code();
-                warn!("connection error: {err} ({err:?}) (code {error_code:?})");
+                debug!("Connection error: {err} ({err:?}) (code {error_code:?})");
 
                 // let's put something useful in debug data
                 let additional_debug_data = format!("{err}").into_bytes();
diff --git a/crates/fluke/src/util.rs b/crates/fluke/src/util.rs
index 898c0835..b3aef610 100644
--- a/crates/fluke/src/util.rs
+++ b/crates/fluke/src/util.rs
@@ -12,6 +12,7 @@ pub(crate) async fn read_and_parse<Parser, Output>(
     stream: &mut impl ReadOwned,
     mut buf: RollMut,
     max_len: usize,
+    // TODO: proper error handling, no eyre::Result
 ) -> eyre::Result<Option<(RollMut, Output)>>
 where
     Parser: Fn(Roll) -> IResult<Roll, Output>,


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index dda97bad..7fbe7209 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -23,7 +23,7 @@ jobs:
       - name: Install Rust specified toolchain
         run: rustup show
       - name: Run sccache-cache
-        uses: mozilla-actions/sccache-action@v0.0.3
+        uses: mozilla-actions/sccache-action@v0.0.4
       - name: Make sure things build
         run: |
           cargo build --profile ci
@@ -42,7 +42,7 @@ jobs:
       - name: Install Rust specified toolchain
         run: rustup show
       - name: Run sccache-cache
-        uses: mozilla-actions/sccache-action@v0.0.3
+        uses: mozilla-actions/sccache-action@v0.0.4
       - name: Make sure things build
         run: |
           cargo build --profile ci
@@ -67,7 +67,7 @@ jobs:
       - name: Install Rust specified toolchain
         run: rustup show
       - name: Run sccache-cache
-        uses: mozilla-actions/sccache-action@v0.0.3
+        uses: mozilla-actions/sccache-action@v0.0.4
       - name: Set up sccache (part 2)
         run: sccache --start-server
       - uses: taiki-e/install-action@v2
diff --git a/test-crates/fluke-h2spec/src/main.rs b/test-crates/fluke-h2spec/src/main.rs
index 4d98579b..139c83eb 100644
--- a/test-crates/fluke-h2spec/src/main.rs
+++ b/test-crates/fluke-h2spec/src/main.rs
@@ -133,8 +133,9 @@ pub(crate) async fn run_server(ln: TcpListener) -> color_eyre::Result<()> {
 
         tokio::task::spawn_local(async move {
             if let Err(e) = fluke::h2::serve(stream.into_halves(), conf, client_buf, driver).await {
-                tracing::error!("error serving client {addr}: {e}, {e:?}");
+                tracing::debug!("error serving client {addr}: {e}, {e:?}");
             }
+            tracing::debug!("done serving client {addr}");
         });
     }
 }