1use super::*;
2use std::{net::TcpListener, thread::JoinHandle, time::Instant};
3
4fn request(stream: &mut TcpStream) -> io::Result<(String, Vec<u8>)> {
5 let head = ws::head(stream)?;
6 let length = ws::header(&head, "Content-Length")
7 .and_then(|length| length.parse().ok())
8 .unwrap_or(0);
9 let mut body = vec![0; length];
10 stream.read_exact(&mut body)?;
11 Ok((head, body))
12}
13
14fn answer(stream: &mut TcpStream, status: u16, body: &[u8]) -> io::Result<()> {
15 write!(
16 stream,
17 "HTTP/1.1 {status} Test\r\nContent-Length: {}\r\n\r\n",
18 body.len()
19 )?;
20 stream.write_all(body)
21}
22
23fn polling<T: Send + 'static>(
24 serve: impl FnOnce(TcpListener, TcpStream) -> T + Send + 'static,
25) -> (Connection, JoinHandle<T>) {
26 proxy::use_proxy(Some(None));
27 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
28 let address = parse(&format!("ws://{}", listener.local_addr().unwrap())).unwrap();
29 let serving = thread::spawn(move || {
30 let (mut stream, _) = listener.accept().unwrap();
31 stream
32 .set_read_timeout(Some(Duration::from_secs(2)))
33 .unwrap();
34 let (head, _) = request(&mut stream).unwrap();
35 assert!(head.starts_with("GET /v1/room/test?poll=1 "));
36 answer(&mut stream, 200, b"session test").unwrap();
37 serve(listener, stream)
38 });
39 let connection = match poll(&address, "/v1/room/test") {
40 Ok(connection) => connection,
41 Err(_) => panic!("the local polling relay refused the session"),
42 };
43 (connection, serving)
44}
45
46fn accept(listener: &TcpListener) -> Option<TcpStream> {
47 listener.set_nonblocking(true).unwrap();
48 let until = Instant::now() + Duration::from_millis(300);
49 loop {
50 match listener.accept() {
51 Ok((stream, _)) => {
52 stream.set_nonblocking(false).unwrap();
53 stream
54 .set_read_timeout(Some(Duration::from_secs(2)))
55 .unwrap();
56 return Some(stream);
57 }
58 Err(error) if error.kind() == io::ErrorKind::WouldBlock && Instant::now() < until => {
59 thread::sleep(Duration::from_millis(1));
60 }
61 Err(error) if error.kind() == io::ErrorKind::WouldBlock => return None,
62 Err(error) => panic!("could not accept the polling request: {error}"),
63 }
64 }
65}
66
67#[test]
68fn an_accepted_post_with_a_lost_answer_retires_its_stream_without_replay() {
69 let sent = ws::frame(ws::BINARY, b"sealed edit", Some([1, 2, 3, 4]));
70 let expected = sent.clone();
71 let (mut connection, serving) = polling(move |listener, mut reading| {
72 assert!(
73 request(&mut reading)
74 .unwrap()
75 .0
76 .starts_with("GET /v1/poll/test ")
77 );
78 let mut posting = accept(&listener).expect("the edit never reached the relay");
79 let (head, body) = request(&mut posting).unwrap();
80 assert!(head.starts_with("POST /v1/poll/test "));
81 assert_eq!(body, expected);
82 drop(posting);
83 let replay = accept(&listener).map(|mut stream| {
84 let (_, body) = request(&mut stream).unwrap();
85 answer(&mut stream, 200, b"").unwrap();
86 body
87 });
88 reading
89 .set_read_timeout(Some(Duration::from_millis(300)))
90 .unwrap();
91 let closed = reading.read(&mut [0]).is_ok_and(|length| length == 0);
92 let _ = answer(&mut reading, 410, b"");
93 (replay, closed)
94 });
95 connection.writer.write_all(&sent).unwrap();
96 let read = connection.reader.read(&mut [0]);
97 let writing_closed = connection.writer.write(&sent).is_err();
98 (connection.close)();
99 let (replay, reading_closed) = serving.join().unwrap();
100 assert!(
101 replay.is_none(),
102 "the same sealed bytes were replayed after an unknown outcome"
103 );
104 assert!(reading_closed, "the other tracked connection stayed open");
105 assert!(read.is_err() || read.is_ok_and(|length| length == 0));
106 assert!(writing_closed);
107}
108
109#[test]
110fn a_drained_get_with_a_truncated_answer_closes_both_tracked_connections() {
111 let (mut connection, serving) = polling(|listener, mut reading| {
112 assert!(
113 request(&mut reading)
114 .unwrap()
115 .0
116 .starts_with("GET /v1/poll/test ")
117 );
118 let mut posting = accept(&listener).expect("the outgoing connection never opened");
119 assert!(
120 request(&mut posting)
121 .unwrap()
122 .0
123 .starts_with("POST /v1/poll/test ")
124 );
125 reading
126 .write_all(b"HTTP/1.1 200 Test\r\nContent-Length: 5\r\n\r\nxx")
127 .unwrap();
128 drop(reading);
129 posting
130 .set_read_timeout(Some(Duration::from_millis(300)))
131 .unwrap();
132 let closed = posting.read(&mut [0]).is_ok_and(|length| length == 0);
133 let _ = answer(&mut posting, 200, b"");
134 let retry = accept(&listener).map(|mut stream| {
135 assert!(
136 request(&mut stream)
137 .unwrap()
138 .0
139 .starts_with("GET /v1/poll/test ")
140 );
141 answer(&mut stream, 200, b"later").unwrap();
142 });
143 (retry.is_some(), closed)
144 });
145 connection
146 .writer
147 .write_all(&ws::frame(ws::PING, b"", Some([0; 4])))
148 .unwrap();
149 let started = Instant::now();
150 let read = connection.reader.read(&mut [0; 5]);
151 let elapsed = started.elapsed();
152 let writing_closed = read.is_err() && connection.writer.write(b"x").is_err();
153 (connection.close)();
154 let (retried, posting_closed) = serving.join().unwrap();
155 assert!(
156 read.is_err(),
157 "bytes consumed by the failed GET were silently discarded"
158 );
159 assert!(
160 !retried,
161 "a failed GET reused the existing encrypted stream"
162 );
163 assert!(posting_closed, "the other tracked connection stayed open");
164 assert!(writing_closed);
165 assert!(elapsed < Duration::from_secs(1));
166}
167
168#[test]
169fn repeated_get_failures_return_without_a_reconnect_loop() {
170 let (mut connection, serving) = polling(|listener, mut reading| {
171 assert!(
172 request(&mut reading)
173 .unwrap()
174 .0
175 .starts_with("GET /v1/poll/test ")
176 );
177 drop(reading);
178 let mut attempts = 1;
179 while let Some(mut stream) = accept(&listener) {
180 assert!(
181 request(&mut stream)
182 .unwrap()
183 .0
184 .starts_with("GET /v1/poll/test ")
185 );
186 attempts += 1;
187 if attempts == 4 {
188 answer(&mut stream, 410, b"").unwrap();
189 break;
190 }
191 }
192 attempts
193 });
194 let started = Instant::now();
195 let read = connection.reader.read(&mut [0]);
196 let elapsed = started.elapsed();
197 (connection.close)();
198 let attempts = serving.join().unwrap();
199 assert!(
200 read.is_err(),
201 "request failures were retried inside the same session"
202 );
203 assert_eq!(attempts, 1);
204 assert!(elapsed < Duration::from_secs(1));
205}