1use super::*;
2use std::{net::TcpListener, thread::JoinHandle};
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}
206
207#[test]
208fn later_resolved_addresses_can_connect_after_the_first_refuses() {
209 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
210 let at = listener.local_addr().unwrap();
211 let refused = SocketAddr::from(([127, 0, 0, 1], 0));
212 let connected = connect_resolved(
213 [refused, at].into_iter(),
214 Instant::now() + Duration::from_secs(2),
215 TcpStream::connect_timeout,
216 );
217 let mut client = match connected {
218 Ok(client) => client,
219 Err((trouble, error)) => panic!("later address failed: {trouble:?}: {error}"),
220 };
221 assert_eq!(client.peer_addr().unwrap(), at);
222 client
223 .set_read_timeout(Some(Duration::from_secs(2)))
224 .unwrap();
225 client
226 .set_write_timeout(Some(Duration::from_secs(2)))
227 .unwrap();
228 let mut server = accept(&listener).expect("the second address never connected");
229 server
230 .set_write_timeout(Some(Duration::from_secs(2)))
231 .unwrap();
232 client.write_all(b"request").unwrap();
233 let mut request = [0; 7];
234 server.read_exact(&mut request).unwrap();
235 assert_eq!(&request, b"request");
236 server.write_all(b"answer").unwrap();
237 let mut answer = [0; 6];
238 client.read_exact(&mut answer).unwrap();
239 assert_eq!(&answer, b"answer");
240}
241
242#[test]
243fn resolved_connection_failures_keep_the_last_error_and_share_the_budget() {
244 let targets = [
245 SocketAddr::from(([127, 0, 0, 1], 1)),
246 SocketAddr::from(([127, 0, 0, 1], 2)),
247 SocketAddr::from(([127, 0, 0, 1], 3)),
248 ];
249 let budget = Duration::from_secs(10);
250 for (last, expected) in [
251 (io::ErrorKind::AddrNotAvailable, Trouble::Unreachable),
252 (io::ErrorKind::TimedOut, Trouble::TimedOut),
253 ] {
254 let mut attempts = 0;
255 let failed = connect_resolved(
256 targets.into_iter(),
257 Instant::now() + budget,
258 |target, timeout| {
259 assert_eq!(*target, targets[attempts]);
260 assert!(!timeout.is_zero());
261 assert!(timeout <= budget / (3 - attempts as u32));
262 attempts += 1;
263 Err(io::Error::new(
264 if attempts == 3 {
265 last
266 } else {
267 io::ErrorKind::ConnectionRefused
268 },
269 format!("attempt {attempts}"),
270 ))
271 },
272 );
273 let (trouble, error) = failed.unwrap_err();
274 assert_eq!(attempts, 3);
275 assert_eq!(trouble, expected);
276 assert_eq!(error.kind(), last);
277 assert_eq!(error.to_string(), "attempt 3");
278 }
279}
280
281#[test]
282fn empty_resolution_and_expired_deadlines_do_not_attempt_a_connection() {
283 let empty = connect_resolved(std::iter::empty(), Instant::now() + CONNECT, |_, _| {
284 panic!("empty resolution tried to connect")
285 });
286 let (trouble, error) = empty.unwrap_err();
287 assert_eq!(trouble, Trouble::Dns);
288 assert_eq!(error.kind(), io::ErrorKind::NotFound);
289 let expired = connect_resolved(
290 [SocketAddr::from(([127, 0, 0, 1], 0))].into_iter(),
291 Instant::now(),
292 |_, _| panic!("an expired deadline tried to connect"),
293 );
294 let (trouble, error) = expired.unwrap_err();
295 assert_eq!(trouble, Trouble::TimedOut);
296 assert_eq!(error.kind(), io::ErrorKind::TimedOut);
297}
298
299#[test]
300fn proxy_tunnels_bracket_ipv6_authorities_and_remain_usable() {
301 for (url, target) in [
302 ("ws://[::1]:123", "[::1]:123"),
303 ("wss://[2001:db8::1]:8443", "[2001:db8::1]:8443"),
304 ("ws://[::1]", "[::1]:80"),
305 ("wss://[::1]", "[::1]:443"),
306 ("ws://127.0.0.1:123", "127.0.0.1:123"),
307 ("ws://127.0.0.1", "127.0.0.1:80"),
308 ("wss://relay.example", "relay.example:443"),
309 ] {
310 let address = parse(url).unwrap();
311 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
312 let at = listener.local_addr().unwrap();
313 let proxy = Proxy {
314 host: "127.0.0.1".into(),
315 port: at.port(),
316 credentials: None,
317 };
318 let mut client = TcpStream::connect_timeout(&at, Duration::from_secs(2)).unwrap();
319 client
320 .set_read_timeout(Some(Duration::from_secs(2)))
321 .unwrap();
322 client
323 .set_write_timeout(Some(Duration::from_secs(2)))
324 .unwrap();
325 let serving = thread::spawn(move || {
326 let mut stream = accept(&listener).expect("the proxy connection never arrived");
327 stream
328 .set_write_timeout(Some(Duration::from_secs(2)))
329 .unwrap();
330 let (head, body) = request(&mut stream).unwrap();
331 assert_eq!(
332 head,
333 format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n")
334 );
335 assert!(body.is_empty());
336 answer(&mut stream, 200, b"").unwrap();
337 let (head, body) = request(&mut stream).unwrap();
338 assert_eq!(head, "GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n");
339 assert!(body.is_empty());
340 answer(&mut stream, 200, b"session test").unwrap();
341 });
342 let exchange = if tunnel(&client, &proxy, &address).is_ok() {
343 (|| -> io::Result<_> {
344 client.write_all(b"GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n")?;
345 request(&mut client)
346 })()
347 } else {
348 Err(io::Error::other("the proxy refused the tunnel"))
349 };
350 serving.join().unwrap();
351 let (head, body) = exchange.unwrap();
352 assert!(head.starts_with("HTTP/1.1 200 "));
353 assert_eq!(body, b"session test");
354 }
355}