grpc_core/http2/
receive.rs

1//! Incremental framing, peer settings and receive-credit ownership.
2use nexora_proto::{
3    ProtoError,
4    http2::{Http2Frame, decode_frame},
5};
6
7use super::*;
8
9impl Http2Session {
10    /// Feeds wire bytes, retaining incomplete prefaces, frames and header blocks.
11    /// # Errors
12    /// Connection errors are sticky; malformed streams produce stream-local events.
13    pub fn mem_recv(&mut self, bytes: &[u8]) -> Result<usize> {
14        self.check_live()?;
15        self.input.extend_from_slice(bytes);
16        if let Err(failure) = self.receive() {
17            self.failed = Some(failure.clone());
18            return Err(failure);
19        }
20        Ok(bytes.len())
21    }
22    fn receive(&mut self) -> Result<()> {
23        while self.preface < CLIENT_PREFACE.len() && !self.input.is_empty() {
24            if self.input.as_slice()[0] != CLIENT_PREFACE[self.preface] {
25                return Err(error(1, "invalid client preface"));
26            }
27            self.preface += 1;
28            self.input.advance(1);
29        }
30        if self.preface < CLIENT_PREFACE.len() {
31            return Ok(());
32        }
33        loop {
34            let raw = self.input.as_slice();
35            if raw.len() < 9 {
36                break;
37            }
38            let length =
39                (usize::from(raw[0]) << 16) | (usize::from(raw[1]) << 8) | usize::from(raw[2]);
40            let kind = raw[3];
41            let flags = raw[4];
42            let id = (u32::from_be_bytes(raw[5..9].try_into().expect("frame header")) & 0x7fff_ffff)
43                as i32;
44            if !self.received_settings && (kind != 4 || flags & 1 != 0) {
45                return Err(error(1, "first frame must be non-ACK SETTINGS"));
46            }
47            if self
48                .headers
49                .as_ref()
50                .is_some_and(|h| kind != 9 || id != h.id)
51            {
52                return Err(error(1, "interleaved header continuation"));
53            }
54            if length > self.local.max_frame_size as usize {
55                return Err(error(6, "frame exceeds local maximum"));
56            }
57            if raw.len() < 9 + length {
58                break;
59            }
60            let settings = if kind == 4 && flags & 1 == 0 {
61                Some(raw[9..9 + length].to_vec())
62            } else {
63                None
64            };
65            let reset = if kind == 3 && length == 4 {
66                Some(u32::from_be_bytes(
67                    raw[9..13].try_into().expect("reset code"),
68                ))
69            } else {
70                None
71            };
72            let goaway = if kind == 7 && length >= 8 {
73                Some(u32::from_be_bytes(
74                    raw[13..17].try_into().expect("shutdown code"),
75                ))
76            } else {
77                None
78            };
79            let frame = match decode_frame(&mut self.input, self.local.max_frame_size) {
80                Ok(Some(frame)) => frame,
81                Ok(None) => break,
82                Err(ProtoError::Http2StreamError {
83                    stream_id,
84                    code,
85                    reason,
86                }) => {
87                    self.input.advance(9 + length);
88                    self.stream_failure(stream_id as i32, code.as_u32() == 11, reason);
89                    continue;
90                }
91                Err(ProtoError::Http2ConnectionError { code, reason }) => {
92                    return Err(error(code.as_u32(), reason));
93                }
94                Err(failure) => return Err(error(1, failure.to_string())),
95            };
96            match frame {
97                Http2Frame::Settings { ack, .. } => {
98                    if ack {
99                        if !self.pending_ack {
100                            return Err(error(1, "unexpected SETTINGS acknowledgement"));
101                        }
102                        self.pending_ack = false;
103                    } else {
104                        self.apply_settings(settings.as_deref().expect("settings payload"))?;
105                        self.received_settings = true;
106                        self.queue_frame(0, FrameType::Settings, 1, &[]);
107                    }
108                }
109                Http2Frame::Headers {
110                    stream_id,
111                    end_stream,
112                    end_headers,
113                    header_block,
114                } => {
115                    self.begin_headers(
116                        stream_id as i32,
117                        end_stream,
118                        header_block.as_slice(),
119                        end_headers,
120                    )?;
121                }
122                Http2Frame::Continuation {
123                    stream_id,
124                    end_headers,
125                    header_block,
126                } => {
127                    let Some(block) = &mut self.headers else {
128                        return Err(error(1, "unexpected continuation"));
129                    };
130                    if block.id != stream_id as i32 {
131                        return Err(error(1, "wrong continuation stream"));
132                    }
133                    block.bytes.extend_from_slice(header_block.as_slice());
134                    self.check_header_block_size()?;
135                    if end_headers {
136                        self.finish_headers()?;
137                    }
138                }
139                Http2Frame::Data {
140                    stream_id,
141                    end_stream,
142                    data,
143                } => {
144                    self.receive_data(stream_id as i32, end_stream, data.as_slice(), length)?;
145                }
146                Http2Frame::WindowUpdate {
147                    stream_id,
148                    increment,
149                } => {
150                    let window = if stream_id == 0 {
151                        &mut self.send_window
152                    } else if let Some(stream) = self.streams.get_mut(&(stream_id as i32)) {
153                        if stream.close_emitted {
154                            continue;
155                        }
156                        &mut stream.send_window
157                    } else if self.idle(stream_id as i32) {
158                        return Err(error(1, "window update on idle stream"));
159                    } else {
160                        continue;
161                    };
162                    if *window + i64::from(increment) > i64::from(i32::MAX) {
163                        if stream_id == 0 {
164                            return Err(error(3, "connection send window overflow"));
165                        }
166                        self.stream_failure(stream_id as i32, false, "stream send window overflow");
167                        continue;
168                    }
169                    *window += i64::from(increment);
170                }
171                Http2Frame::RstStream { stream_id, .. } => {
172                    let id = stream_id as i32;
173                    if self.idle(id) {
174                        return Err(error(1, "reset on idle stream"));
175                    }
176                    if let Some(stream) = self.streams.get_mut(&id)
177                        && !stream.close_emitted
178                    {
179                        stream.close_emitted = true;
180                        stream.reset = true;
181                        stream.provider = None;
182                        self.events.push_back(SessionEvent::StreamClose {
183                            stream_id: id,
184                            error_code: reset.expect("reset payload"),
185                        });
186                    }
187                }
188                Http2Frame::GoAway { last_stream_id, .. } => {
189                    let last = self
190                        .received_goaway
191                        .map_or(last_stream_id as i32, |old| old.min(last_stream_id as i32));
192                    self.received_goaway = Some(last);
193                    self.events.push_back(SessionEvent::GoAway {
194                        last_stream_id: last,
195                        error_code: goaway.expect("shutdown payload"),
196                    });
197                    for (&id, stream) in &mut self.streams {
198                        if self.role == Http2Role::Client && id > last && !stream.close_emitted {
199                            stream.reset = true;
200                            stream.close_emitted = true;
201                            stream.provider = None;
202                            self.events.push_back(SessionEvent::StreamClose {
203                                stream_id: id,
204                                error_code: 7,
205                            });
206                        }
207                    }
208                }
209                Http2Frame::Ping {
210                    ack: false,
211                    opaque_data,
212                } => self.queue_frame(0, FrameType::Ping, 1, &opaque_data),
213                Http2Frame::Unknown { header, .. }
214                    if header.frame_type == FrameType::PushPromise =>
215                {
216                    return Err(error(1, "server push is disabled"));
217                }
218                _ => {}
219            }
220        }
221        // A fully drained connection can close without emitting queued control acknowledgements.
222        if self.received_goaway.is_some() && self.streams.values().all(|s| s.close_emitted) {
223            self.outgoing.retain(|item| {
224                !matches!(
225                    item,
226                    Outgoing::Frame {
227                        id: 0,
228                        kind: FrameType::Settings,
229                        flags: 1,
230                        ..
231                    }
232                )
233            });
234        }
235        Ok(())
236    }
237    fn idle(&self, id: i32) -> bool {
238        if id <= 0 {
239            return true;
240        }
241        if id & 1 == 0 {
242            return true;
243        }
244        if self.role == Http2Role::Client {
245            id >= self.next_id
246        } else {
247            id > self.last_peer
248        }
249    }
250    fn apply_settings(&mut self, payload: &[u8]) -> Result<()> {
251        for setting in payload.as_chunks::<6>().0 {
252            let id = u16::from_be_bytes(setting[..2].try_into().expect("setting ID"));
253            let value = u32::from_be_bytes(setting[2..].try_into().expect("setting value"));
254            match id {
255                1 => {
256                    self.peer.header_table_size = value;
257                    self.encoder.set_max_table_size(value.min(4096) as usize);
258                }
259                2 => {
260                    if value > 1 || self.role == Http2Role::Client {
261                        return Err(error(1, "invalid ENABLE_PUSH setting"));
262                    }
263                    self.peer.enable_push = value != 0;
264                }
265                3 => self.peer.max_concurrent_streams = value,
266                4 => {
267                    if value > i32::MAX as u32 {
268                        return Err(error(3, "invalid initial window"));
269                    }
270                    let delta = i64::from(value) - i64::from(self.peer.initial_window_size);
271                    if self
272                        .streams
273                        .values()
274                        .any(|s| !s.close_emitted && s.send_window + delta > i64::from(i32::MAX))
275                    {
276                        return Err(error(3, "initial window update overflow"));
277                    }
278                    for stream in self.streams.values_mut().filter(|s| !s.close_emitted) {
279                        stream.send_window += delta;
280                    }
281                    self.peer.initial_window_size = value;
282                }
283                5 => {
284                    if !(16_384..=16_777_215).contains(&value) {
285                        return Err(error(1, "invalid frame size setting"));
286                    }
287                    self.peer.max_frame_size = value;
288                }
289                6 => self.peer.max_header_list_size = value,
290                _ => {}
291            }
292        }
293        Ok(())
294    }
295    fn receive_data(&mut self, id: i32, end: bool, data: &[u8], wire_len: usize) -> Result<()> {
296        self.recv_window -= wire_len as i64;
297        if self.recv_window < 0 {
298            return Err(error(3, "connection receive window exceeded"));
299        }
300        let Some(stream) = self.streams.get_mut(&id) else {
301            if self.idle(id) {
302                return Err(error(1, "DATA on idle stream"));
303            }
304            self.credit_connection(wire_len)?;
305            return Ok(());
306        };
307        if stream.reset || stream.close_emitted {
308            self.credit_connection(wire_len)?;
309            return Ok(());
310        }
311        stream.recv_window -= wire_len as i64;
312        if stream.recv_window < 0 || !stream.initial || stream.remote_closed {
313            self.credit_connection(wire_len)?;
314            self.stream_failure(id, false, "DATA violates stream state or flow window");
315            return Ok(());
316        }
317        stream.unconsumed += data.len();
318        if end {
319            stream.remote_closed = true;
320        }
321        self.events.push_back(SessionEvent::DataChunk {
322            stream_id: id,
323            data: data.to_vec(),
324        });
325        self.events.push_back(SessionEvent::FrameRecv {
326            stream_id: id,
327            frame_type: FrameType::Data,
328            flags: u8::from(end),
329        });
330        let padding = wire_len - data.len();
331        self.credit_connection(padding)?;
332        self.credit_stream(id, padding)?;
333        self.close_if_complete(id);
334        Ok(())
335    }
336    /// Returns application-consumed bytes to both receive windows in bounded batches.
337    /// # Errors
338    /// Rejects unknown streams and credit exceeding delivered DATA bytes.
339    pub fn consume(&mut self, id: i32, amount: usize) -> Result<()> {
340        let stream = self
341            .streams
342            .get_mut(&id)
343            .ok_or_else(|| error(1, "consume on unknown stream"))?;
344        if amount > stream.unconsumed {
345            return Err(error(1, "receive credit exceeds delivered bytes"));
346        }
347        stream.unconsumed -= amount;
348        self.credit_connection(amount)?;
349        self.credit_stream(id, amount)
350    }
351    pub(super) fn credit_connection(&mut self, amount: usize) -> Result<()> {
352        self.credit = self
353            .credit
354            .checked_add(u32::try_from(amount).map_err(|_| error(3, "receive credit overflow"))?)
355            .ok_or_else(|| error(3, "receive credit overflow"))?;
356        if self.credit >= 32_768 {
357            let credit = std::mem::take(&mut self.credit);
358            self.recv_window += i64::from(credit);
359            self.queue_frame(0, FrameType::WindowUpdate, 0, &credit.to_be_bytes());
360        }
361        Ok(())
362    }
363    fn credit_stream(&mut self, id: i32, amount: usize) -> Result<()> {
364        let stream = self.streams.get_mut(&id).expect("delivered stream");
365        if stream.reset || stream.remote_closed {
366            return Ok(());
367        }
368        stream.credit = stream
369            .credit
370            .checked_add(u32::try_from(amount).map_err(|_| error(3, "stream credit overflow"))?)
371            .ok_or_else(|| error(3, "stream credit overflow"))?;
372        if stream.credit >= (self.local.initial_window_size / 2).max(1) {
373            let credit = std::mem::take(&mut stream.credit);
374            stream.recv_window += i64::from(credit);
375            self.queue_frame(id, FrameType::WindowUpdate, 0, &credit.to_be_bytes());
376        }
377        Ok(())
378    }
379}
grpc_core/http2/receive.rs

grpc_core/http2/
receive.rs
