conmonrs/streaming_server/spdy/
reader.rs

1//! Reads the frames of the client and hands them to the streams.
2
3use super::{
4    Connection, Live, SHUTDOWN_TIMEOUT, STREAM_QUEUE, Shared, Stream, StreamReceiver, StreamSender,
5    StreamState,
6    frame::{
7        Decoder, Frame, GOAWAY_PROTOCOL_ERROR, Headers, STATUS_CANCEL, STATUS_PROTOCOL_ERROR,
8        STATUS_REFUSED_STREAM,
9    },
10};
11use anyhow::{Result, bail};
12use std::{
13    io,
14    sync::{
15        Arc,
16        atomic::{AtomicBool, Ordering},
17    },
18};
19use tokio::{
20    io::{AsyncRead, AsyncReadExt, ReadHalf},
21    select,
22    sync::mpsc,
23    time,
24};
25use tracing::{debug, trace, warn};
26
27/// Reads the frames of the client and dispatches them to the streams.
28pub(super) struct Reader<T, F> {
29    pub(super) shared: Arc<Shared>,
30    pub(super) reader: ReadHalf<T>,
31    pub(super) accept: F,
32    pub(super) streams: mpsc::Sender<Stream>,
33}
34
35/// Why the reader stopped.
36enum End {
37    /// The client closed the connection, or the connection broke.
38    ClientClosed,
39
40    /// Nothing takes streams anymore or the connection ended otherwise, like
41    /// when it was idle.
42    Closed,
43
44    /// The client violated the protocol, the connection cannot continue.
45    ProtocolError(anyhow::Error),
46}
47
48impl<T, F> Reader<T, F>
49where
50    T: AsyncRead,
51    F: Fn(&Headers) -> Result<()>,
52{
53    /// Read the frames of the client until the connection ended, then end
54    /// the streams.
55    pub(super) async fn run(mut self) {
56        let shared = self.shared.clone();
57        let end = self.read().await;
58        if let End::ProtocolError(e) = &end {
59            warn!("Closing SPDY connection: {e:#}");
60            shared.go_away(GOAWAY_PROTOCOL_ERROR);
61        }
62
63        shared.done.cancel();
64        // Ends the data of all streams, the reset tokens are children of
65        // `done` and got cancelled with it.
66        shared.lock_streams().clear();
67
68        // Nothing gets accepted anymore.
69        drop(self.streams);
70        if !matches!(end, End::ClientClosed) {
71            Self::drain(&mut self.reader).await;
72        }
73    }
74
75    /// Read the client after the end of the connection until it closed it
76    /// as well, for up to [`SHUTDOWN_TIMEOUT`]. A socket closed with input
77    /// it did not read resets the connection, which can drop the last
78    /// frames of the server, like the status of a session.
79    async fn drain(reader: &mut ReadHalf<T>) {
80        let mut buf = [0; 4096];
81        let drained = async { while matches!(reader.read(&mut buf).await, Ok(1..)) {} };
82        if time::timeout(SHUTDOWN_TIMEOUT, drained).await.is_err() {
83            debug!("SPDY client did not close the connection in time");
84        }
85    }
86
87    async fn read(&mut self) -> End {
88        let mut decoder = Decoder::default();
89        let mut going_away = false;
90
91        loop {
92            let frame = select! {
93                frame = decoder.read(&mut self.reader) => frame,
94                () = self.shared.done.cancelled() => return End::Closed,
95            };
96            let frame = match frame {
97                Ok(Some(frame)) => frame,
98                Ok(None) => return End::ClientClosed,
99                // A connection which broke, also within a frame, is a
100                // client which went away and not one which violated the
101                // protocol: the kubelet resets the connection when its own
102                // client went away with output it did not read. No GOAWAY
103                // reaches it anymore.
104                Err(e) if e.chain().any(|cause| cause.is::<io::Error>()) => {
105                    debug!("SPDY client connection ended: {e:#}");
106                    return End::ClientClosed;
107                }
108                Err(e) => return End::ProtocolError(e),
109            };
110            if Connection::is_activity(&frame) {
111                self.shared.activity.touch();
112            }
113            trace!("Got SPDY frame {frame:?}");
114
115            let result = match frame {
116                Frame::SynStream {
117                    stream_id,
118                    fin,
119                    headers,
120                } => {
121                    // Client stream IDs are odd and increase.
122                    let last_stream_id = self.shared.last_stream_id.load(Ordering::Relaxed);
123                    if stream_id % 2 == 0 || stream_id <= last_stream_id {
124                        self.shared
125                            .send_control(Frame::RstStream {
126                                stream_id,
127                                status: STATUS_PROTOCOL_ERROR,
128                            })
129                            .await
130                    } else {
131                        self.shared
132                            .last_stream_id
133                            .store(stream_id, Ordering::Relaxed);
134                        self.open(stream_id, fin, headers, going_away).await
135                    }
136                }
137                Frame::Data {
138                    stream_id,
139                    fin,
140                    data,
141                } => {
142                    self.shared.deliver(stream_id, data).await;
143                    if fin {
144                        self.shared.remote_closed(stream_id);
145                    }
146                    Ok(())
147                }
148                Frame::Headers { stream_id, fin } => {
149                    if fin {
150                        self.shared.remote_closed(stream_id);
151                    }
152                    Ok(())
153                }
154                Frame::RstStream { stream_id, status } => {
155                    debug!("Client reset SPDY stream {stream_id} with status {status}");
156                    self.shared.remove(stream_id);
157                    Ok(())
158                }
159                // Pings of the client have odd IDs and get echoed, even ones are
160                // the echoes of the pings of the server.
161                Frame::Ping { id } if id % 2 == 1 => {
162                    self.shared.send_control(Frame::Ping { id }).await
163                }
164                Frame::GoAway { .. } => {
165                    debug!("Client is going away");
166                    going_away = true;
167                    Ok(())
168                }
169                Frame::Ping { .. } | Frame::SynReply { .. } | Frame::Ignored => Ok(()),
170            };
171            if result.is_err() {
172                return End::Closed;
173            }
174        }
175    }
176
177    /// Accept or refuse a stream the client opened.
178    #[expect(
179        clippy::needless_pass_by_ref_mut,
180        reason = "a shared reference would need the reader to be Sync for a Send future"
181    )]
182    async fn open(
183        &mut self,
184        stream_id: u32,
185        fin: bool,
186        headers: Headers,
187        going_away: bool,
188    ) -> Result<()> {
189        let refused = if going_away || self.streams.is_closed() {
190            Some(STATUS_REFUSED_STREAM)
191        } else if !self.shared.has_room() {
192            debug!("Refusing SPDY stream {stream_id}, too many streams are open");
193            Some(STATUS_REFUSED_STREAM)
194        } else if let Err(e) = (self.accept)(&headers) {
195            debug!("Rejecting SPDY stream {stream_id}: {e:#}");
196            Some(STATUS_CANCEL)
197        } else {
198            None
199        };
200        if let Some(status) = refused {
201            return self
202                .shared
203                .send_control(Frame::RstStream { stream_id, status })
204                .await;
205        }
206
207        let reset = self.shared.done.child_token();
208        let remote_closed = Arc::new(AtomicBool::new(fin));
209        let (data_tx, data_rx) = mpsc::channel(STREAM_QUEUE);
210        self.shared.lock_streams().insert(
211            stream_id,
212            StreamState {
213                data: (!fin).then_some(data_tx),
214                reset: reset.clone(),
215                remote_closed: remote_closed.clone(),
216                local_closed: false,
217            },
218        );
219        // Queued before anything the stream could send.
220        self.shared
221            .send_control(Frame::SynReply {
222                stream_id,
223                headers: Headers::default(),
224            })
225            .await?;
226
227        let live = Live::new(&self.shared);
228        let stream = Stream {
229            headers,
230            receiver: StreamReceiver {
231                data: data_rx,
232                reset: reset.clone(),
233                remote_closed,
234                _live: live.clone(),
235            },
236            sender: StreamSender {
237                id: stream_id,
238                shared: self.shared.clone(),
239                reset,
240                queued: Arc::default(),
241                closed: false,
242                _live: live,
243            },
244        };
245        select! {
246            // Dropping a stream nobody accepts anymore resets it.
247            _ = self.streams.send(stream) => Ok(()),
248            () = self.shared.done.cancelled() => bail!("connection closed"),
249        }
250    }
251}
conmonrs/streaming_server/spdy/reader.rs

conmonrs/streaming_server/spdy/
reader.rs
